From dcde64ce407b332072525c66885adf03a26eeb5e Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Wed, 12 Oct 2016 11:07:01 +0100 Subject: [PATCH 1/2] Fix import order. Fix typos and improve readability --- .ci/flake8_lint_include_list.txt | 1 + lib/galaxy/workflow/extract.py | 87 ++++++++++++++-------------- test/api/test_workflow_extraction.py | 4 +- 3 files changed, 45 insertions(+), 47 deletions(-) diff --git a/.ci/flake8_lint_include_list.txt b/.ci/flake8_lint_include_list.txt index eb2ab916d61..343e3818220 100644 --- a/.ci/flake8_lint_include_list.txt +++ b/.ci/flake8_lint_include_list.txt @@ -347,6 +347,7 @@ lib/galaxy/web/framework/middleware/statsd.py lib/galaxy/web/framework/middleware/translogger.py lib/galaxy/web/framework/middleware/xforwardedhost.py lib/galaxy/web/params.py +lib/galaxy/workflow/extract.py lib/galaxy/workflow/__init__.py lib/galaxy/workflow/render.py lib/galaxy/workflow/schedulers/__init__.py diff --git a/lib/galaxy/workflow/extract.py b/lib/galaxy/workflow/extract.py index e9469a02767..adc61d0e1c9 100644 --- a/lib/galaxy/workflow/extract.py +++ b/lib/galaxy/workflow/extract.py @@ -1,25 +1,26 @@ """ This module contains functionality to aid in extracting workflows from histories. """ -from galaxy.util.odict import odict -from galaxy import exceptions -from galaxy import model +import logging + +from galaxy import exceptions, model from galaxy.tools.parameters.basic import ( - DataToolParameter, - DataCollectionToolParameter + DataCollectionToolParameter, + DataToolParameter ) -from galaxy.tools.parser import ToolOutputCollectionPart from galaxy.tools.parameters.grouping import ( Conditional, Repeat, Section ) +from galaxy.tools.parser import ToolOutputCollectionPart +from galaxy.util.odict import odict + from .steps import ( attach_ordered_steps, order_workflow_steps_with_levels ) -import logging log = logging.getLogger( __name__ ) WARNING_SOME_DATASETS_NOT_READY = "Some datasets still queued or running were ignored" @@ -67,16 +68,14 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_ elif type( dataset_collection_ids) is not list: dataset_collection_ids = [ dataset_collection_ids ] # Convert both sets of ids to integers - job_ids = [ int( id ) for id in job_ids ] - dataset_ids = [ int( id ) for id in dataset_ids ] - dataset_collection_ids = [ int( id ) for id in dataset_collection_ids ] - # Find each job, for security we (implicately) check that they are - # associated witha job in the current history. + job_ids = [ int( _ ) for _ in job_ids ] + dataset_ids = [ int( _ ) for _ in dataset_ids ] + dataset_collection_ids = [ int( _ ) for _ in dataset_collection_ids ] + # Find each job, for security we (implicitly) check that they are + # associated with a job in the current history. summary = WorkflowSummary( trans, history ) jobs = summary.jobs - jobs_by_id = dict( ( job.id, job ) for job in jobs.keys() ) steps = [] - steps_by_job_id = {} hid_to_output_pair = {} # Input dataset steps for i, hid in enumerate( dataset_ids ): @@ -103,6 +102,7 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_ hid_to_output_pair[ hid ] = ( step, 'output' ) steps.append( step ) # Tool steps + jobs_by_id = dict( ( job.id, job ) for job in jobs.keys() ) for job_id in job_ids: if job_id not in jobs_by_id: log.warning( "job_id %s not found in jobs_by_id %s" % ( job_id, jobs_by_id ) ) @@ -134,7 +134,6 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_ conn.output_step = other_step conn.output_name = other_name steps.append( step ) - steps_by_job_id[ job_id ] = step # Store created dataset hids for assoc in (job.output_datasets + job.output_dataset_collection_instances): assoc_name = assoc.name @@ -147,8 +146,8 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_ if query_assoc_name == assoc_name: hid = dataset_collection.hid if hid is None: - template = "Failed to find matching implicit job - job is %s, jobs are %s, assoc_name is %s." - message = template % ( job.id, jobs, assoc.name ) + template = "Failed to find matching implicit job - job id is %s, implicit pairs are %s, assoc_name is %s." + message = template % ( job.id, jobs[job], assoc_name ) log.warning( message ) raise Exception( "Failed to extract job." ) else: @@ -223,25 +222,24 @@ class WorkflowSummary( object ): else: self.__summarize_dataset( content ) - def __summarize_dataset_collection( self, content ): - content = self.__original_hdca( content ) - dataset_collection = content - hid = content.hid - self.collection_types[ hid ] = content.collection.collection_type - cja = content.creating_job_associations + def __summarize_dataset_collection( self, dataset_collection ): + dataset_collection = self.__original_hdca( dataset_collection ) + hid = dataset_collection.hid + self.collection_types[ hid ] = dataset_collection.collection.collection_type + cja = dataset_collection.creating_job_associations if cja: - # Use the first job to represent all mapped jobs. - representive_job_assoc = content.creating_job_associations[0] - job = representive_job_assoc.job - if job not in self.jobs or self.jobs[ job ][ 0 ][ 1 ].history_content_type == "dataset": - self.jobs[ job ] = [ ( representive_job_assoc.name, dataset_collection ) ] - if content.implicit_output_name: - self.implicit_map_jobs.append( job ) + # Use the "first" job to represent all mapped jobs. + representative_assoc = cja[0] + representative_job = representative_assoc.job + if representative_job not in self.jobs or self.jobs[ representative_job ][ 0 ][ 1 ].history_content_type == "dataset": + self.jobs[ representative_job ] = [ ( representative_assoc.name, dataset_collection ) ] + if dataset_collection.implicit_output_name: + self.implicit_map_jobs.append( representative_job ) else: - self.jobs[ job ].append( ( representive_job_assoc.name, dataset_collection ) ) + self.jobs[ representative_job ].append( ( representative_assoc.name, dataset_collection ) ) # This whole elif condition may no longer be needed do to additional # tracking with creating_job_associations. Will delete at some point. - elif content.implicit_output_name: + elif dataset_collection.implicit_output_name: # TODO: Optimize db call dataset_instance = dataset_collection.collection.dataset_instances[ 0 ] if not self.__check_state( dataset_instance ): @@ -249,13 +247,13 @@ class WorkflowSummary( object ): # makes me wonder if even need this check at all? return - job_hda = self.__original_hda( dataset_instance ) - if not job_hda.creating_job_associations: + original_hda = self.__original_hda( dataset_instance ) + if not original_hda.creating_job_associations: log.warning( "An implicitly create output dataset collection doesn't have a creating_job_association, should not happen!" ) job = DatasetCollectionCreationJob( dataset_collection ) self.jobs[ job ] = [ ( None, dataset_collection ) ] - for assoc in job_hda.creating_job_associations: + for assoc in original_hda.creating_job_associations: job = assoc.job if job not in self.jobs or self.jobs[ job ][ 0 ][ 1 ].history_content_type == "dataset": self.jobs[ job ] = [ ( assoc.name, dataset_collection ) ] @@ -263,19 +261,19 @@ class WorkflowSummary( object ): else: self.jobs[ job ].append( ( assoc.name, dataset_collection ) ) else: - job = DatasetCollectionCreationJob( content ) - self.jobs[ job ] = [ ( None, content ) ] + job = DatasetCollectionCreationJob( dataset_collection ) + self.jobs[ job ] = [ ( None, dataset_collection ) ] def __summarize_dataset( self, dataset ): if not self.__check_state( dataset ): return - job_hda = self.__original_hda( dataset ) + original_hda = self.__original_hda( dataset ) - if not job_hda.creating_job_associations: + if not original_hda.creating_job_associations: self.jobs[ FakeJob( dataset ) ] = [ ( None, dataset ) ] - for assoc in job_hda.creating_job_associations: + for assoc in original_hda.creating_job_associations: job = assoc.job if job in self.jobs: self.jobs[ job ].append( ( assoc.name, dataset ) ) @@ -288,11 +286,10 @@ class WorkflowSummary( object ): return hdca def __original_hda( self, hda ): - # if this hda was copied from another, we need to find the job that created the origial hda - job_hda = hda - while job_hda.copied_from_history_dataset_association: - job_hda = job_hda.copied_from_history_dataset_association - return job_hda + # if this hda was copied from another, we need to find the job that created the original hda + while hda.copied_from_history_dataset_association: + hda = hda.copied_from_history_dataset_association + return hda def __check_state( self, hda ): # FIXME: Create "Dataset.is_finished" diff --git a/test/api/test_workflow_extraction.py b/test/api/test_workflow_extraction.py index 41545bbe3e4..53864393edd 100644 --- a/test/api/test_workflow_extraction.py +++ b/test/api/test_workflow_extraction.py @@ -216,7 +216,7 @@ test_data: text_input2: "samp1\t30.0\nsamp2\t40.0\n" """) tool_ids = [ "cat1", "collection_split_on_column", "cat_list" ] - job_ids = list(map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids )) + job_ids = [ functools.partial(self._job_id_for_tool, jobs_summary.jobs )(_) for _ in tool_ids ] downloaded_workflow = self._extract_and_download_workflow( dataset_ids=[ "1", "2" ], job_ids=job_ids, @@ -266,7 +266,7 @@ test_data: content: "samp1\t30.0\nsamp2\t40.0\n" """) tool_ids = [ "cat1", "collection_creates_pair", "cat_collection", "cat_list" ] - job_ids = list(map( functools.partial(self._job_id_for_tool, jobs_summary.jobs ), tool_ids )) + job_ids = [ functools.partial(self._job_id_for_tool, jobs_summary.jobs )(_) for _ in tool_ids ] downloaded_workflow = self._extract_and_download_workflow( dataset_collection_ids=[ "3" ], job_ids=job_ids, From 04bf2153b8f61c60288cb180d448e7c20352f6e7 Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Wed, 12 Oct 2016 14:01:29 +0100 Subject: [PATCH 2/2] WorkflowSummary: add dict to map a job id to its representative job Fix #2742 introduced in commit 9ede8e556be1f6038361b579c1457068e50e951a . The representative job is the first in the returned query, so substantially random and not necessarily the one with the lowest job id. Mapping a job id to its representative job in extract_steps() makes it robust also against API requests with manually chosen collection job ids. --- lib/galaxy/workflow/extract.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/lib/galaxy/workflow/extract.py b/lib/galaxy/workflow/extract.py index adc61d0e1c9..4b3a3bc9289 100644 --- a/lib/galaxy/workflow/extract.py +++ b/lib/galaxy/workflow/extract.py @@ -102,12 +102,11 @@ def extract_steps( trans, history=None, job_ids=None, dataset_ids=None, dataset_ hid_to_output_pair[ hid ] = ( step, 'output' ) steps.append( step ) # Tool steps - jobs_by_id = dict( ( job.id, job ) for job in jobs.keys() ) for job_id in job_ids: - if job_id not in jobs_by_id: - log.warning( "job_id %s not found in jobs_by_id %s" % ( job_id, jobs_by_id ) ) + if job_id not in summary.job_id2representative_job: + log.warning( "job_id %s not found in job_id2representative_job %s" % ( job_id, summary.job_id2representative_job ) ) raise AssertionError( "Attempt to create workflow with job not connected to current history" ) - job = jobs_by_id[ job_id ] + job = summary.job_id2representative_job[job_id] tool_inputs, associations = step_inputs( trans, job ) step = model.WorkflowStep() step.type = 'tool' @@ -201,6 +200,7 @@ class WorkflowSummary( object ): self.history = history self.warnings = set() self.jobs = odict() + self.job_id2representative_job = {} # map a non-fake job id to its representative job self.implicit_map_jobs = [] self.collection_types = {} @@ -237,6 +237,9 @@ class WorkflowSummary( object ): self.implicit_map_jobs.append( representative_job ) else: self.jobs[ representative_job ].append( ( representative_assoc.name, dataset_collection ) ) + for assoc in cja: + job = assoc.job + self.job_id2representative_job[job.id] = representative_job # This whole elif condition may no longer be needed do to additional # tracking with creating_job_associations. Will delete at some point. elif dataset_collection.implicit_output_name: @@ -257,6 +260,7 @@ class WorkflowSummary( object ): job = assoc.job if job not in self.jobs or self.jobs[ job ][ 0 ][ 1 ].history_content_type == "dataset": self.jobs[ job ] = [ ( assoc.name, dataset_collection ) ] + self.job_id2representative_job[job.id] = job self.implicit_map_jobs.append( job ) else: self.jobs[ job ].append( ( assoc.name, dataset_collection ) ) @@ -279,6 +283,7 @@ class WorkflowSummary( object ): self.jobs[ job ].append( ( assoc.name, dataset ) ) else: self.jobs[ job ] = [ ( assoc.name, dataset ) ] + self.job_id2representative_job[job.id] = job def __original_hdca( self, hdca ): while hdca.copied_from_history_dataset_collection_association: