From 069c604aa822381ae8b122da3fefb30e74372d24 Mon Sep 17 00:00:00 2001 From: "Michael R. Crusoe" Date: Wed, 10 Nov 2021 16:36:09 +0100 Subject: [PATCH] improve typing of lib/galaxy/tools/execute.py --- lib/galaxy/tools/execute.py | 22 +++++++++++++++++----- setup.cfg | 2 -- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index d45419b7535..69eb89ae804 100644 --- a/lib/galaxy/tools/execute.py +++ b/lib/galaxy/tools/execute.py @@ -5,6 +5,8 @@ collections from matched collections. """ import collections import logging +from abc import abstractmethod +from typing import Dict, List from boltons.iterutils import remap @@ -44,7 +46,9 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle ) if invocation_step is None: - execution_tracker = ToolExecutionTracker(trans, tool, mapping_params, collection_info, completed_jobs=completed_jobs) + execution_tracker: ExecutionTracker = ToolExecutionTracker( + trans, tool, mapping_params, collection_info, completed_jobs=completed_jobs + ) else: execution_tracker = WorkflowStepExecutionTracker(trans, tool, mapping_params, collection_info, invocation_step, completed_jobs=completed_jobs) execution_cache = ToolExecutionCache(trans) @@ -100,7 +104,9 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle jobs_executed = 0 has_remaining_jobs = False execution_slice = None - job_datasets = {} # job: list of dataset instances created by job + job_datasets: Dict[ + str, List[model.DatasetInstance] + ] = {} # job: list of dataset instances created by job for i, execution_slice in enumerate(execution_tracker.new_execution_slices()): if max_num_jobs is not None and jobs_executed >= max_num_jobs: @@ -122,10 +128,12 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle trans.sa_session.flush() tool_id = tool.id - for job in execution_tracker.successful_jobs: + for job2 in execution_tracker.successful_jobs: # Put the job in the queue if tracking in memory - tool.app.job_manager.enqueue(job, tool=tool, flush=False) - trans.log_event(f"Added job to the job queue, id: {str(job.id)}", tool_id=tool_id) + tool.app.job_manager.enqueue(job2, tool=tool, flush=False) + trans.log_event( + f"Added job to the job queue, id: {str(job2.id)}", tool_id=tool_id + ) trans.sa_session.flush() if has_remaining_jobs: @@ -409,6 +417,10 @@ class ExecutionTracker: job_assoc.job = job self.trans.sa_session.add(job_assoc) + @abstractmethod + def new_collection_execution_slices(self): + pass + # Seperate these because workflows need to track their jobs belong to the invocation # in the database immediately and they can be recovered. diff --git a/setup.cfg b/setup.cfg index 99f300c7dde..3e0a5aa6beb 100644 --- a/setup.cfg +++ b/setup.cfg @@ -557,8 +557,6 @@ check_untyped_defs = False check_untyped_defs = False [mypy-galaxy.managers.ratable] check_untyped_defs = False -[mypy-galaxy.tools.execute] -check_untyped_defs = False [mypy-galaxy.tools.actions.upload] check_untyped_defs = False [mypy-galaxy.tools.actions.model_operations]