Source code for galaxy.tools.actions.metadata
import logging
import os
from json import dumps
from typing import (
Any,
Dict,
Optional,
)
from galaxy.job_execution.datasets import DatasetPath
from galaxy.metadata import get_metadata_compute_strategy
from galaxy.model import (
History,
Job,
User,
)
from galaxy.model.base import transaction
from galaxy.model.dataset_collections.matching import MatchingCollections
from galaxy.tools._types import ToolStateJobInstancePopulatedT
from galaxy.tools.execute import (
DatasetCollectionElementsSliceT,
DEFAULT_DATASET_COLLECTION_ELEMENTS,
DEFAULT_JOB_CALLBACK,
DEFAULT_PREFERRED_OBJECT_STORE_ID,
DEFAULT_RERUN_REMAP_JOB_ID,
DEFAULT_SET_OUTPUT_HID,
JobCallbackT,
)
from galaxy.tools.execution_helpers import ToolExecutionCache
from galaxy.util import asbool
from . import ToolAction
log = logging.getLogger(__name__)
[docs]class SetMetadataToolAction(ToolAction):
"""Tool action used for setting external metadata on an existing dataset"""
produces_real_jobs: bool = False
set_output_hid: bool = False
[docs] def execute(
self,
tool,
trans,
incoming: Optional[ToolStateJobInstancePopulatedT] = None,
history: Optional[History] = None,
job_params=None,
rerun_remap_job_id: Optional[int] = DEFAULT_RERUN_REMAP_JOB_ID,
execution_cache: Optional[ToolExecutionCache] = None,
dataset_collection_elements: Optional[DatasetCollectionElementsSliceT] = DEFAULT_DATASET_COLLECTION_ELEMENTS,
completed_job: Optional[Job] = None,
collection_info: Optional[MatchingCollections] = None,
job_callback: Optional[JobCallbackT] = DEFAULT_JOB_CALLBACK,
preferred_object_store_id: Optional[str] = DEFAULT_PREFERRED_OBJECT_STORE_ID,
set_output_hid: bool = DEFAULT_SET_OUTPUT_HID,
flush_job: bool = True,
skip: bool = False,
):
"""
Execute using a web transaction.
"""
overwrite = True
job, odict = self.execute_via_trans(
tool,
trans,
incoming,
overwrite,
history,
job_params,
)
# FIXME: can remove this when logging in execute_via_app method.
trans.log_event(f"Added set external metadata job to the job queue, id: {str(job.id)}", tool_id=job.tool_id)
return job, odict
[docs] def execute_via_trans(
self,
tool,
trans,
incoming: Optional[Dict[str, Any]],
overwrite: bool = True,
history: Optional[History] = None,
job_params: Optional[Dict[str, Any]] = None,
):
trans.check_user_activation()
session = trans.get_galaxy_session()
session_id = session and session.id
history_id = trans.history and trans.history.id
incoming = incoming or {}
return self.execute_via_app(
tool,
trans.app,
session_id,
history_id,
trans.user,
incoming,
overwrite,
history,
job_params,
)
[docs] def execute_via_app(
self,
tool,
app,
session_id: Optional[int],
history_id: Optional[int],
user: Optional[User] = None,
incoming: Optional[Dict[str, Any]] = None,
overwrite: bool = True,
history: Optional[History] = None,
job_params: Optional[Dict[str, Any]] = None,
):
"""
Execute using application.
"""
incoming = incoming or {}
for name, value in incoming.items():
# Why are we looping here and not just using a fixed input name? Needed?
if not name.startswith("input"):
continue
if isinstance(value, app.model.HistoryDatasetAssociation):
dataset = value
dataset_name = name
type = "hda"
break
elif isinstance(value, app.model.LibraryDatasetDatasetAssociation):
dataset = value
dataset_name = name
type = "ldda"
break
else:
raise Exception("The dataset to set metadata on could not be determined.")
sa_session = app.model.context
# Create the job object
job = app.model.Job()
job.galaxy_version = app.config.version_major
job.session_id = session_id
job.history_id = history_id
job.tool_id = tool.id
if user:
job.user_id = user.id
if job_params:
job.params = dumps(job_params)
start_job_state = job.state # should be job.states.NEW
try:
# For backward compatibility, some tools may not have versions yet.
job.tool_version = tool.version
except AttributeError:
job.tool_version = "1.0.1"
job.dynamic_tool = tool.dynamic_tool
job.state = (
job.states.WAITING
) # we need to set job state to something other than NEW, or else when tracking jobs in db it will be picked up before we have added input / output parameters
sa_session.add(job)
with transaction(sa_session): # ensure job.id is available
sa_session.commit()
# add parameters to job_parameter table
# Store original dataset state, so we can restore it. A separate table might be better (no chance of 'losing' the original state)?
incoming["__ORIGINAL_DATASET_STATE__"] = dataset.state
input_paths = [DatasetPath(dataset.id, real_path=dataset.get_file_name(), mutable=False)]
app.object_store.create(job, base_dir="job_work", dir_only=True, extra_dir=str(job.id))
job_working_dir = app.object_store.get_filename(job, base_dir="job_work", dir_only=True, extra_dir=str(job.id))
datatypes_config = os.path.join(job_working_dir, "registry.xml")
app.datatypes_registry.to_xml_file(path=datatypes_config)
external_metadata_wrapper = get_metadata_compute_strategy(app.config, job.id, tool_id=tool.id)
output_datatasets_dict = {
dataset_name: dataset,
}
validate_outputs = asbool(incoming.get("validate", False))
cmd_line = external_metadata_wrapper.setup_external_metadata(
output_datatasets_dict,
{},
sa_session,
exec_dir=None,
tmp_dir=job_working_dir,
dataset_files_path=app.model.Dataset.file_path,
output_fnames=input_paths,
config_root=app.config.root,
config_file=app.config.config_file,
datatypes_config=datatypes_config,
job_metadata=os.path.join(job_working_dir, "working", tool.provided_metadata_file),
include_command=False,
max_metadata_value_size=app.config.max_metadata_value_size,
max_discovered_files=app.config.max_discovered_files,
validate_outputs=validate_outputs,
job=job,
kwds={"overwrite": overwrite},
)
incoming["__SET_EXTERNAL_METADATA_COMMAND_LINE__"] = cmd_line
for name, value in tool.params_to_strings(incoming, app).items():
job.add_parameter(name, value)
# add the dataset to job_to_input_dataset table
if type == "hda":
job.add_input_dataset(dataset_name, dataset)
elif type == "ldda":
job.add_input_library_dataset(dataset_name, dataset)
# Need a special state here to show that metadata is being set and also allow the job to run
# i.e. if state was set to 'running' the set metadata job would never run, as it would wait for input (the dataset to set metadata on) to be in a ready state
dataset.state = dataset.states.SETTING_METADATA
job.state = start_job_state # job inputs have been configured, restore initial job state
with transaction(sa_session):
sa_session.commit()
# clear e.g. converted files
dataset.datatype.before_setting_metadata(dataset)
return job, {}