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

@@ -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)

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:

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
 

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,

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):