From 48c654d1b26e080fc3d13a96c91dc53182948314 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Thu, 22 Sep 2022 12:23:59 +0200 Subject: [PATCH] Add type hint for galaxy.workflow.{run, run_request} Trying out https://github.com/dropbox/pyannotate here. It's still a lot of work even with pyannotate, but I think this fixes some minor oddities, like assigning non-mapped attributes to WorkflowInvocation instances. --- lib/galaxy/workflow/modules.py | 6 +- lib/galaxy/workflow/run.py | 175 ++++++++++++------ lib/galaxy/workflow/run_request.py | 113 +++++++---- lib/galaxy/workflow/schedulers/core.py | 9 +- test/unit/workflows/test_workflow_progress.py | 2 +- 5 files changed, 205 insertions(+), 100 deletions(-) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index e3dd5e35c7f..63b527f93a5 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -690,7 +690,7 @@ class InputModule(WorkflowModule): # Web controller may set copy_inputs_to_history, API controller always sets # inputs. - if invocation.copy_inputs_to_history: + if progress.copy_inputs_to_history: for input_dataset_hda in list(step_outputs.values()): content_type = input_dataset_hda.history_content_type if content_type == "dataset": @@ -1991,7 +1991,7 @@ class ToolModule(WorkflowModule): invocation_step=invocation_step, max_num_jobs=max_num_jobs, validate_outputs=validate_outputs, - job_callback=lambda job: self._handle_post_job_actions(step, job, invocation.replacement_dict), + job_callback=lambda job: self._handle_post_job_actions(step, job, progress.replacement_dict), completed_jobs=completed_jobs, workflow_resource_parameters=resource_parameters, ) @@ -2015,7 +2015,7 @@ class ToolModule(WorkflowModule): step_inputs = mapping_params.param_template step_inputs.update(collection_info.collections) - self._handle_mapped_over_post_job_actions(step, step_inputs, step_outputs, invocation.replacement_dict) + self._handle_mapped_over_post_job_actions(step, step_inputs, step_outputs, progress.replacement_dict) if execution_tracker.execution_errors: message = "Failed to create one or more job(s) for workflow step." raise Exception(message) diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index ac5af627826..eaa2caa181f 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -1,11 +1,20 @@ import logging import uuid from typing import ( + Any, + Dict, List, + Optional, + Tuple, + TYPE_CHECKING, Union, ) from galaxy import model +from galaxy.model import ( + WorkflowInvocation, + WorkflowInvocationStep, +) from galaxy.util import ExecutionTimer from galaxy.workflow import modules from galaxy.workflow.run_request import ( @@ -14,15 +23,39 @@ from galaxy.workflow.run_request import ( WorkflowRunConfig, ) +if TYPE_CHECKING: + from galaxy.model import ( + Workflow, + WorkflowOutput, + WorkflowStep, + WorkflowStepConnection, + ) + from galaxy.webapps.base.webapp import GalaxyWebTransaction + from galaxy.work.context import WorkRequestContext + from galaxy.workflow.modules import WorkflowModuleInjector + log = logging.getLogger(__name__) +WorkflowOutputsType = Dict[int, Any] + # Entry point for core workflow scheduler. -def schedule(trans, workflow, workflow_run_config, workflow_invocation): +def schedule( + trans: "WorkRequestContext", + workflow: "Workflow", + workflow_run_config: WorkflowRunConfig, + workflow_invocation: WorkflowInvocation, +) -> Tuple[WorkflowOutputsType, WorkflowInvocation]: return __invoke(trans, workflow, workflow_run_config, workflow_invocation) -def __invoke(trans, workflow, workflow_run_config, workflow_invocation=None, populate_state=False): +def __invoke( + trans: "WorkRequestContext", + workflow: "Workflow", + workflow_run_config: WorkflowRunConfig, + workflow_invocation: Optional[WorkflowInvocation] = None, + populate_state: bool = False, +) -> Tuple[WorkflowOutputsType, WorkflowInvocation]: """Run the supplied workflow in the supplied target_history.""" if populate_state: modules.populate_module_and_state( @@ -38,34 +71,35 @@ def __invoke(trans, workflow, workflow_run_config, workflow_invocation=None, pop workflow_run_config, workflow_invocation=workflow_invocation, ) + workflow_invocation = invoker.workflow_invocation try: outputs = invoker.invoke() except modules.CancelWorkflowEvaluation: - if workflow_invocation: - if workflow_invocation.cancel(): - trans.sa_session.add(workflow_invocation) - outputs = [] + if workflow_invocation.cancel(): + trans.sa_session.add(workflow_invocation) + outputs = {} except Exception: log.exception("Failed to execute scheduled workflow.") - if workflow_invocation: - # Running workflow invocation in background, just mark - # persistent workflow invocation as failed. - workflow_invocation.fail() - trans.sa_session.add(workflow_invocation) - else: - # Running new transient workflow invocation in legacy - # controller action - propage the exception up. - raise - outputs = [] + # Running workflow invocation in background, just mark + # persistent workflow invocation as failed. + workflow_invocation.fail() + trans.sa_session.add(workflow_invocation) + outputs = {} - if workflow_invocation: - # Be sure to update state of workflow_invocation. - trans.sa_session.flush() + # Be sure to update state of workflow_invocation. + trans.sa_session.flush() - return outputs, invoker.workflow_invocation + return outputs, workflow_invocation -def queue_invoke(trans, workflow, workflow_run_config, request_params=None, populate_state=True, flush=True): +def queue_invoke( + trans: "GalaxyWebTransaction", + workflow: "Workflow", + workflow_run_config: WorkflowRunConfig, + request_params: Optional[Dict[str, Any]] = None, + populate_state: bool = True, + flush: bool = True, +) -> WorkflowInvocation: request_params = request_params or {} if populate_state: modules.populate_module_and_state( @@ -80,9 +114,17 @@ def queue_invoke(trans, workflow, workflow_run_config, request_params=None, popu class WorkflowInvoker: - def __init__(self, trans, workflow, workflow_run_config, workflow_invocation=None, progress=None): + def __init__( + self, + trans: "WorkRequestContext", + workflow: "Workflow", + workflow_run_config: WorkflowRunConfig, + workflow_invocation: Optional[WorkflowInvocation] = None, + progress: Optional["WorkflowProgress"] = None, + ) -> None: self.trans = trans self.workflow = workflow + self.workflow_invocation: WorkflowInvocation if progress is not None: assert workflow_invocation is None workflow_invocation = progress.workflow_invocation @@ -90,7 +132,7 @@ class WorkflowInvoker: if workflow_invocation is None: invocation_uuid = uuid.uuid1() - workflow_invocation = model.WorkflowInvocation() + workflow_invocation = WorkflowInvocation() workflow_invocation.workflow = self.workflow # In one way or another, following attributes will become persistent @@ -98,13 +140,7 @@ class WorkflowInvoker: workflow_invocation.uuid = invocation_uuid workflow_invocation.history = workflow_run_config.target_history - self.workflow_invocation = workflow_invocation - else: - self.workflow_invocation = workflow_invocation - - self.workflow_invocation.copy_inputs_to_history = workflow_run_config.copy_inputs_to_history - self.workflow_invocation.use_cached_job = workflow_run_config.use_cached_job - self.workflow_invocation.replacement_dict = workflow_run_config.replacement_dict + self.workflow_invocation = workflow_invocation module_injector = modules.WorkflowModuleInjector(trans) if progress is None: @@ -116,10 +152,13 @@ class WorkflowInvoker: jobs_per_scheduling_iteration=getattr( trans.app.config, "maximum_workflow_jobs_per_scheduling_iteration", -1 ), + copy_inputs_to_history=workflow_run_config.copy_inputs_to_history, + use_cached_job=workflow_run_config.use_cached_job, + replacement_dict=workflow_run_config.replacement_dict, ) self.progress = progress - def invoke(self): + def invoke(self) -> Dict[int, Any]: workflow_invocation = self.workflow_invocation config = self.trans.app.config maximum_duration = getattr(config, "maximum_workflow_invocation_duration", -1) @@ -153,13 +192,15 @@ class WorkflowInvoker: self.__check_implicitly_dependent_steps(step) if not workflow_invocation_step: - workflow_invocation_step = model.WorkflowInvocationStep() + workflow_invocation_step = WorkflowInvocationStep() + assert workflow_invocation_step workflow_invocation_step.workflow_invocation = workflow_invocation workflow_invocation_step.workflow_step = step workflow_invocation_step.state = "new" workflow_invocation.steps.append(workflow_invocation_step) + assert workflow_invocation_step incomplete_or_none = self._invoke_step(workflow_invocation_step) if incomplete_or_none is False: step_delayed = delayed_steps = True @@ -228,9 +269,9 @@ class WorkflowInvoker: if job.state != job.states.OK: raise modules.CancelWorkflowEvaluation() - def _invoke_step(self, invocation_step): + def _invoke_step(self, invocation_step: WorkflowInvocationStep) -> Optional[bool]: incomplete_or_none = invocation_step.workflow_step.module.execute( - self.trans, self.progress, invocation_step, use_cached_job=self.workflow_invocation.use_cached_job + self.trans, self.progress, invocation_step, use_cached_job=self.progress.use_cached_job ) return incomplete_or_none @@ -240,27 +281,40 @@ STEP_OUTPUT_DELAYED = object() class WorkflowProgress: def __init__( - self, workflow_invocation, inputs_by_step_id, module_injector, param_map, jobs_per_scheduling_iteration=-1 - ): - self.outputs = {} + self, + workflow_invocation: WorkflowInvocation, + inputs_by_step_id: Any, + module_injector: "WorkflowModuleInjector", + param_map: Dict[int, Dict[str, Any]], + jobs_per_scheduling_iteration: int = -1, + copy_inputs_to_history: bool = False, + use_cached_job: bool = False, + replacement_dict: Optional[Dict[str, str]] = None, + ) -> None: + self.outputs: Dict[int, Any] = {} self.module_injector = module_injector self.workflow_invocation = workflow_invocation self.inputs_by_step_id = inputs_by_step_id self.param_map = param_map self.jobs_per_scheduling_iteration = jobs_per_scheduling_iteration self.jobs_scheduled_this_iteration = 0 + self.copy_inputs_to_history = copy_inputs_to_history + self.use_cached_job = use_cached_job + self.replacement_dict = replacement_dict or {} @property - def maximum_jobs_to_schedule_or_none(self): + def maximum_jobs_to_schedule_or_none(self) -> Optional[int]: if self.jobs_per_scheduling_iteration > 0: return self.jobs_per_scheduling_iteration - self.jobs_scheduled_this_iteration else: return None - def record_executed_job_count(self, job_count): + def record_executed_job_count(self, job_count: int) -> None: self.jobs_scheduled_this_iteration += job_count - def remaining_steps(self): + def remaining_steps( + self, + ) -> List[Tuple["WorkflowStep", Optional[WorkflowInvocationStep]]]: # Previously computed and persisted step states. step_states = self.workflow_invocation.step_states_by_step_id() steps = self.workflow_invocation.workflow.steps @@ -287,7 +341,7 @@ class WorkflowProgress: remaining_steps.append((step, invocation_step)) return remaining_steps - def replacement_for_input(self, step, input_dict): + def replacement_for_input(self, step: "WorkflowStep", input_dict: Dict[str, Any]) -> Any: replacement: Union[ modules.NoReplacement, model.DatasetCollectionInstance, @@ -315,7 +369,7 @@ class WorkflowProgress: return replacement - def replacement_for_connection(self, connection, is_data=True): + def replacement_for_connection(self, connection: "WorkflowStepConnection", is_data: bool = True) -> Any: output_step_id = connection.output_step.id if output_step_id not in self.outputs: message = f"No outputs found for step id {output_step_id}, outputs are {self.outputs}" @@ -368,7 +422,7 @@ class WorkflowProgress: return replacement - def get_replacement_workflow_output(self, workflow_output): + def get_replacement_workflow_output(self, workflow_output: "WorkflowOutput") -> Any: step = workflow_output.workflow_step output_name = workflow_output.output_name step_outputs = self.outputs[step.id] @@ -378,7 +432,9 @@ class WorkflowProgress: else: return step_outputs[output_name] - def set_outputs_for_input(self, invocation_step, outputs=None, already_persisted=False): + def set_outputs_for_input( + self, invocation_step: WorkflowInvocationStep, outputs: Any = None, already_persisted: bool = False + ) -> None: step = invocation_step.workflow_step if outputs is None: @@ -399,7 +455,9 @@ class WorkflowProgress: self.set_step_outputs(invocation_step, outputs, already_persisted=already_persisted) - def set_step_outputs(self, invocation_step, outputs, already_persisted=False): + def set_step_outputs( + self, invocation_step: WorkflowInvocationStep, outputs: Dict[str, Any], already_persisted: bool = False + ) -> None: step = invocation_step.workflow_step if invocation_step.output_value: outputs[invocation_step.output_value.workflow_output.output_name] = invocation_step.output_value.value @@ -435,25 +493,27 @@ class WorkflowProgress: output=output, ) - def _record_workflow_output(self, step, workflow_output, output): + def _record_workflow_output(self, step: "WorkflowStep", workflow_output: "WorkflowOutput", output: Any) -> None: self.workflow_invocation.add_output(workflow_output, step, output) - def mark_step_outputs_delayed(self, step, why=None): + def mark_step_outputs_delayed(self, step: "WorkflowStep", why: Optional[str] = None) -> None: if why: message = f"Marking step {step.id} outputs of invocation {self.workflow_invocation.id} delayed ({why})" log.debug(message) self.outputs[step.id] = STEP_OUTPUT_DELAYED - def _subworkflow_invocation(self, step): + def _subworkflow_invocation(self, step: "WorkflowStep") -> WorkflowInvocation: workflow_invocation = self.workflow_invocation subworkflow_invocation = workflow_invocation.get_subworkflow_invocation_for_step(step) if subworkflow_invocation is None: raise Exception(f"Failed to find persisted workflow invocation for step [{step.id}]") return subworkflow_invocation - def subworkflow_invoker(self, trans, step, use_cached_job=False): + def subworkflow_invoker( + self, trans: "WorkRequestContext", step: "WorkflowStep", use_cached_job: bool = False + ) -> WorkflowInvoker: subworkflow_invocation = self._subworkflow_invocation(step) - workflow_run_config = workflow_request_to_run_config(trans, subworkflow_invocation) + workflow_run_config = workflow_request_to_run_config(subworkflow_invocation, use_cached_job) subworkflow_progress = self.subworkflow_progress(subworkflow_invocation, step, workflow_run_config.param_map) subworkflow_invocation = subworkflow_progress.workflow_invocation return WorkflowInvoker( @@ -463,7 +523,9 @@ class WorkflowProgress: progress=subworkflow_progress, ) - def subworkflow_progress(self, subworkflow_invocation, step, param_map): + def subworkflow_progress( + self, subworkflow_invocation: WorkflowInvocation, step: "WorkflowStep", param_map: Dict + ) -> "WorkflowProgress": subworkflow = subworkflow_invocation.workflow subworkflow_inputs = {} for input_subworkflow_step in subworkflow.input_steps: @@ -483,9 +545,16 @@ class WorkflowProgress: if not connection_found: raise Exception("Could not find connections for all subworkflow inputs.") - return WorkflowProgress(subworkflow_invocation, subworkflow_inputs, self.module_injector, param_map=param_map) + return WorkflowProgress( + subworkflow_invocation, + subworkflow_inputs, + self.module_injector, + param_map=param_map, + use_cached_job=self.use_cached_job, + replacement_dict=self.replacement_dict, + ) - def _recover_mapping(self, step_invocation): + def _recover_mapping(self, step_invocation: WorkflowInvocationStep) -> None: try: step_invocation.workflow_step.module.recover_mapping(step_invocation, self) except modules.DelayedWorkflowEvaluation as de: diff --git a/lib/galaxy/workflow/run_request.py b/lib/galaxy/workflow/run_request.py index 6cb858ac948..6272d6fa393 100644 --- a/lib/galaxy/workflow/run_request.py +++ b/lib/galaxy/workflow/run_request.py @@ -1,15 +1,35 @@ import json import logging import uuid -from typing import Dict +from typing import ( + Any, + Dict, + List, + Optional, + TYPE_CHECKING, +) -from galaxy import ( - exceptions, - model, +from galaxy import exceptions +from galaxy.model import ( + Dataset, + History, + HistoryDatasetAssociation, + LibraryDataset, + LibraryDatasetDatasetAssociation, + WorkflowInvocation, + WorkflowRequestInputParameter, + WorkflowRequestStepState, ) from galaxy.tools.parameters.meta import expand_workflow_inputs from galaxy.workflow.resources import get_resource_mapper_function +if TYPE_CHECKING: + from galaxy.model import ( + Workflow, + WorkflowStep, + ) + from galaxy.webapps.base.webapp import GalaxyWebTransaction + INPUT_STEP_TYPES = ["data_input", "data_collection_input", "parameter_input"] log = logging.getLogger(__name__) @@ -45,17 +65,17 @@ class WorkflowRunConfig: def __init__( self, - target_history, - replacement_dict, - copy_inputs_to_history=False, - inputs=None, - param_map=None, - allow_tool_state_corrections=False, - use_cached_job=False, - resource_params=None, - ): + target_history: "History", + replacement_dict: Optional[Dict[str, Any]] = None, + inputs: Optional[Dict[int, Any]] = None, + param_map: Optional[Dict[int, Any]] = None, + allow_tool_state_corrections: bool = False, + copy_inputs_to_history: bool = False, + use_cached_job: bool = False, + resource_params: Optional[Dict[int, Any]] = None, + ) -> None: self.target_history = target_history - self.replacement_dict = replacement_dict + self.replacement_dict = replacement_dict or {} self.copy_inputs_to_history = copy_inputs_to_history self.inputs = inputs or {} self.param_map = param_map or {} @@ -64,7 +84,9 @@ class WorkflowRunConfig: self.use_cached_job = use_cached_job -def _normalize_inputs(steps, inputs, inputs_by): +def _normalize_inputs( + steps: List["WorkflowStep"], inputs: Dict[str, Dict[str, Any]], inputs_by: str +) -> Dict[int, Dict[str, Any]]: normalized_inputs = {} for step in steps: if step.type not in INPUT_STEP_TYPES: @@ -101,7 +123,9 @@ def _normalize_inputs(steps, inputs, inputs_by): return normalized_inputs -def _normalize_step_parameters(steps, param_map, legacy=False, already_normalized=False): +def _normalize_step_parameters( + steps: List["WorkflowStep"], param_map: Dict, legacy: bool = False, already_normalized: bool = False +) -> Dict: """Take a complex param_map that can reference parameters by step_id in the new flexible way or in the old one-parameter per step fashion or by tool id and normalize the parameters so @@ -132,7 +156,7 @@ def _normalize_step_parameters(steps, param_map, legacy=False, already_normalize return normalized_param_map -def _step_parameters(step, param_map, legacy=False): +def _step_parameters(step: "WorkflowStep", param_map: Dict, legacy: bool = False) -> Dict: """ Update ``step`` parameters based on the user-provided ``param_map`` dict. @@ -177,7 +201,7 @@ def _step_parameters(step, param_map, legacy=False): return new_params -def _flatten_step_params(param_dict, prefix=""): +def _flatten_step_params(param_dict: Dict, prefix: str = "") -> Dict: # TODO: Temporary work around until tool code can process nested data # structures. This should really happen in there so the tools API gets # this functionality for free and so that repeats can be handled @@ -198,7 +222,13 @@ def _flatten_step_params(param_dict, prefix=""): return new_params -def _get_target_history(trans, workflow, payload, param_keys=None, index=0): +def _get_target_history( + trans: "GalaxyWebTransaction", + workflow: "Workflow", + payload: Dict[str, Any], + param_keys: Optional[List[List]] = None, + index: int = 0, +) -> History: param_keys = param_keys or [] history_name = payload.get("new_history_name", None) history_id = payload.get("history_id", None) @@ -230,14 +260,16 @@ def _get_target_history(trans, workflow, payload, param_keys=None, index=0): nh_name = f"{nh_name} on {ids[0]}" elif nids > 1: nh_name = f"{nh_name} on {', '.join(ids[0:-1])} and {ids[-1]}" - new_history = trans.app.model.History(user=trans.user, name=nh_name) + new_history = History(user=trans.user, name=nh_name) trans.sa_session.add(new_history) trans.sa_session.flush() target_history = new_history return target_history -def build_workflow_run_configs(trans, workflow, payload): +def build_workflow_run_configs( + trans: "GalaxyWebTransaction", workflow: "Workflow", payload: Dict[str, Any] +) -> List[WorkflowRunConfig]: app = trans.app allow_tool_state_corrections = payload.get("allow_tool_state_corrections", False) use_cached_job = payload.get("use_cached_job", False) @@ -327,7 +359,7 @@ def build_workflow_run_configs(trans, workflow, payload): input_id = input_dict["id"] try: if input_source == "ldda": - ldda = trans.sa_session.query(app.model.LibraryDatasetDatasetAssociation).get( + ldda = trans.sa_session.query(LibraryDatasetDatasetAssociation).get( trans.security.decode_id(input_id) ) assert trans.user_is_admin or trans.app.security_agent.can_access_dataset( @@ -336,7 +368,7 @@ def build_workflow_run_configs(trans, workflow, payload): content = ldda.to_history_dataset_association(history, add_to_history=add_to_history) elif input_source == "ld": ldda = ( - trans.sa_session.query(app.model.LibraryDataset) + trans.sa_session.query(LibraryDataset) .get(trans.security.decode_id(input_id)) .library_dataset_dataset_association ) @@ -346,16 +378,12 @@ def build_workflow_run_configs(trans, workflow, payload): content = ldda.to_history_dataset_association(history, add_to_history=add_to_history) elif input_source == "hda": # Get dataset handle, add to dict and history if necessary - content = trans.sa_session.query(app.model.HistoryDatasetAssociation).get( - trans.security.decode_id(input_id) - ) + content = trans.sa_session.query(HistoryDatasetAssociation).get(trans.security.decode_id(input_id)) assert trans.user_is_admin or trans.app.security_agent.can_access_dataset( trans.get_current_user_roles(), content.dataset ) elif input_source == "uuid": - dataset = ( - trans.sa_session.query(app.model.Dataset).filter(app.model.Dataset.uuid == input_id).first() - ) + dataset = trans.sa_session.query(Dataset).filter(Dataset.uuid == input_id).first() if dataset is None: # this will need to be changed later. If federation code is avalible, then a missing UUID # could be found amoung fereration partners @@ -428,15 +456,17 @@ def build_workflow_run_configs(trans, workflow, payload): return run_configs -def workflow_run_config_to_request(trans, run_config, workflow): - param_types = model.WorkflowRequestInputParameter.types +def workflow_run_config_to_request( + trans: "GalaxyWebTransaction", run_config: WorkflowRunConfig, workflow: "Workflow" +) -> WorkflowInvocation: + param_types = WorkflowRequestInputParameter.types - workflow_invocation = model.WorkflowInvocation() + workflow_invocation = WorkflowInvocation() workflow_invocation.uuid = uuid.uuid1() workflow_invocation.history = run_config.target_history - def add_parameter(name, value, type): - parameter = model.WorkflowRequestInputParameter( + def add_parameter(name: str, value: str, type: WorkflowRequestInputParameter.types) -> None: + parameter = WorkflowRequestInputParameter( name=name, value=value, type=type, @@ -448,7 +478,7 @@ def workflow_run_config_to_request(trans, run_config, workflow): steps_by_id[step.id] = step serializable_runtime_state = step.module.encode_runtime_state(step.state) - step_state = model.WorkflowRequestStepState() + step_state = WorkflowRequestStepState() step_state.workflow_step = step log.info(f"Creating a step_state for step.id {step.id}") step_state.value = serializable_runtime_state @@ -461,7 +491,7 @@ def workflow_run_config_to_request(trans, run_config, workflow): copy_inputs_to_history=False, use_cached_job=run_config.use_cached_job, inputs={}, - param_map=run_config.param_map.get(step.order_index, {}), + param_map=run_config.param_map.get(step.order_index), allow_tool_state_corrections=run_config.allow_tool_state_corrections, resource_params=run_config.resource_params, ) @@ -486,14 +516,14 @@ def workflow_run_config_to_request(trans, run_config, workflow): workflow_invocation.add_input(content, step_id) for step_id, param_dict in run_config.param_map.items(): add_parameter( - name=step_id, + name=str(step_id), value=json.dumps(param_dict), type=param_types.STEP_PARAMETERS, ) resource_parameters = run_config.resource_params for key, value in resource_parameters.items(): - add_parameter(key, value, param_types.RESOURCE_PARAMETERS) + add_parameter(str(key), value, param_types.RESOURCE_PARAMETERS) add_parameter( "copy_inputs_to_history", "true" if run_config.copy_inputs_to_history else "false", param_types.META_PARAMETERS ) @@ -501,15 +531,16 @@ def workflow_run_config_to_request(trans, run_config, workflow): return workflow_invocation -def workflow_request_to_run_config(work_request_context, workflow_invocation): - param_types = model.WorkflowRequestInputParameter.types +def workflow_request_to_run_config( + workflow_invocation: WorkflowInvocation, use_cached_job: bool = False +) -> WorkflowRunConfig: + param_types = WorkflowRequestInputParameter.types history = workflow_invocation.history replacement_dict = {} inputs = {} param_map = {} resource_params = {} copy_inputs_to_history = None - use_cached_job = False for parameter in workflow_invocation.input_parameters: parameter_type = parameter.type diff --git a/lib/galaxy/workflow/schedulers/core.py b/lib/galaxy/workflow/schedulers/core.py index a4a51d4f17b..a66c6db44eb 100644 --- a/lib/galaxy/workflow/schedulers/core.py +++ b/lib/galaxy/workflow/schedulers/core.py @@ -2,6 +2,7 @@ it simply schedules the whole workflow up front when offered. """ import logging +from typing import TYPE_CHECKING from galaxy.work import context from galaxy.workflow import ( @@ -10,6 +11,10 @@ from galaxy.workflow import ( ) from ..schedulers import ActiveWorkflowSchedulingPlugin +if TYPE_CHECKING: + from galaxy.model import WorkflowInvocation + + log = logging.getLogger(__name__) @@ -25,13 +30,13 @@ class CoreWorkflowSchedulingPlugin(ActiveWorkflowSchedulingPlugin): def shutdown(self): pass - def schedule(self, workflow_invocation): + def schedule(self, workflow_invocation: "WorkflowInvocation") -> None: workflow = workflow_invocation.workflow history = workflow_invocation.history request_context = context.WorkRequestContext( app=self.app, history=history, user=history.user ) # trans-like object not tied to a web-thread. - workflow_run_config = run_request.workflow_request_to_run_config(request_context, workflow_invocation) + workflow_run_config = run_request.workflow_request_to_run_config(workflow_invocation) run.schedule( trans=request_context, workflow=workflow, diff --git a/test/unit/workflows/test_workflow_progress.py b/test/unit/workflows/test_workflow_progress.py index 3b173c886cf..36ae9346edb 100644 --- a/test/unit/workflows/test_workflow_progress.py +++ b/test/unit/workflows/test_workflow_progress.py @@ -76,7 +76,7 @@ class WorkflowProgressTestCase(unittest.TestCase): self.invocation.workflow = workflow def _new_workflow_progress(self): - return WorkflowProgress(self.invocation, self.inputs_by_step_id, MockModuleInjector(self.progress), {}) + return WorkflowProgress(self.invocation, self.inputs_by_step_id, MockModuleInjector(self.progress), {}) # type: ignore[arg-type] def _set_previous_progress(self, outputs): for i, (step_id, step_value) in enumerate(outputs):