opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-25 00:12:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

Use final_job_state in data manager exec_after_process

Browse Source 
This was all working correctly in my test, but removes some custom logic.
It was theoretically possible for the old logic to fail if there was
no output dataset for the job.
This commit is contained in:
mvdbeek
2019-11-06 14:25:18 +01:00
parent 1b776b5b0b
commit d6b69cf6d5
2 changed files with 7 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/__init__.py
+1 -1
View File
@@ -1638,7 +1638,7 @@ class JobWrapper(HasResourceParameters):
# Certain tools require tasks to be completed after job execution
# ( this used to be performed in the "exec_after_process" hook, but hooks are deprecated ).
param_dict = self.get_param_dict(job)
self.tool.exec_after_process(self.app, inp_data, out_data, param_dict, job=job)
self.tool.exec_after_process(self.app, inp_data, out_data, param_dict, job=job, final_job_state=final_job_state)
# Call 'exec_after_process' hook
self.tool.call_hook('exec_after_process', self.app, inp_data=inp_data,
out_data=out_data, param_dict=param_dict,
lib/galaxy/tools/__init__.py
+6 -9
View File
@@ -43,6 +43,7 @@ from galaxy.tool_util.loader import (
raw_tool_xml_tree,
template_macro_params
)
from galaxy.tool_util.output_checker import DETECTED_JOB_STATE
from galaxy.tool_util.parser import (
get_tool_source,
get_tool_source_from_representation,
@@ -1747,7 +1748,7 @@ class Tool(Dictifiable):
def exec_before_job(self, app, inp_data, out_data, param_dict={}):
pass
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None):
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None, **kwds):
pass
def job_failed(self, job_wrapper, message, exception=False):
@@ -2415,7 +2416,7 @@ class SetMetadataTool(Tool):
history.id, job.user, incoming={'input1': hda}, overwrite=False
)
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None):
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None, **kwds):
working_directory = app.object_store.get_filename(
job, base_dir='job_work', dir_only=True, obj_dir=True
)
@@ -2499,17 +2500,13 @@ class DataManagerTool(OutputParameterJSONTool):
if self.data_manager_id is None:
self.data_manager_id = self.id
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None, **kwds):
def exec_after_process(self, app, inp_data, out_data, param_dict, job=None, final_job_state=None, **kwds):
assert self.allow_user_access(job.user), "You must be an admin to access this tool."
if final_job_state != DETECTED_JOB_STATE.OK:
return
# run original exec_after_process
super(DataManagerTool, self).exec_after_process(app, inp_data, out_data, param_dict, job=job, **kwds)
# process results of tool
if job and job.state == job.states.ERROR:
return
# Job state may now be 'running' instead of previous 'error', but datasets are still set to e.g. error
for dataset in out_data.values():
if dataset.state != dataset.states.OK:
return
data_manager_id = job.data_manager_association.data_manager_id
data_manager = self.app.data_managers.get_manager(data_manager_id, None)
assert data_manager is not None, "Invalid data manager (%s) requested. It may have been removed before the job completed." % (data_manager_id)