Fix various type hints

lib/galaxy/model/__init__.py

@@ -8695,7 +8695,7 @@ class WorkflowInvocation(Base, UsesCreateAndUpdateTime, Dictifiable, Serializabl
     state: Mapped[Optional[str]] = mapped_column(TrimmedString(64), index=True)
     scheduler: Mapped[Optional[str]] = mapped_column(TrimmedString(255), index=True)
     handler: Mapped[Optional[str]] = mapped_column(TrimmedString(255), index=True)
-    uuid: Mapped[Optional[Union[UUID, str]]] = mapped_column(UUIDType())
+    uuid: Mapped[Optional[Union[UUID]]] = mapped_column(UUIDType())
     history_id: Mapped[Optional[int]] = mapped_column(ForeignKey("history.id"), index=True)
 
     history = relationship("History", back_populates="workflow_invocations")

lib/galaxy/model/store/ro_crate_utils.py

@@ -358,6 +358,7 @@ def _add_step_parameter_pv(self, step: WorkflowInvocationStep, crate: ROCrate):
 
     def _add_step_parameter_fp(self, step: WorkflowInvocationStep, crate: ROCrate):
         param_id = step.workflow_step.label
+        assert step.workflow_step.tool_inputs
         param_type = step.workflow_step.tool_inputs["parameter_type"]
         return crate.add(
             ContextEntity(
@@ -375,6 +376,7 @@ def _add_step_parameter_fp(self, step: WorkflowInvocationStep, crate: ROCrate):
 
     def _add_step_tool_pv(self, step: WorkflowInvocationStep, tool_input: str, crate: ROCrate):
         param_id = tool_input
+        assert step.workflow_step.tool_inputs
         return crate.add(
             ContextEntity(
                 crate,

lib/galaxy/workflow/modules.py

@@ -463,7 +463,7 @@ def update_value(input, context, prefixed_name, **kwargs):
 
         return state, step_errors
 
-    def encode_runtime_state(self, step, runtime_state):
+    def encode_runtime_state(self, step, runtime_state: DefaultToolState):
         """Takes the computed runtime state and serializes it during run request creation."""
         return runtime_state.encode(Bunch(inputs=self.get_runtime_inputs(step)), self.trans.app)
 
@@ -2274,6 +2274,7 @@ def execute(
             message = f"Specified tool [{tool.id}] in step {step.order_index + 1} is not workflow-compatible."
             raise exceptions.MessageException(message)
         tool_state = step.state
+        assert tool_state is not None
         tool_inputs = tool.inputs.copy()
         # Not strictly needed - but keep Tool state clean by stripping runtime
         # metadata parameters from it.
@@ -2402,7 +2403,7 @@ def callback(input, prefixed_name: str, **kwargs):
                 mapping_params=mapping_params,
                 history=invocation.history,
                 collection_info=collection_info,
-                workflow_invocation_uuid=invocation.uuid.hex,
+                workflow_invocation_uuid=invocation.uuid.hex if invocation.uuid else None,
                 invocation_step=invocation_step,
                 max_num_jobs=max_num_jobs,
                 validate_outputs=validate_outputs,

lib/galaxy/workflow/run.py

@@ -339,6 +339,7 @@ def __check_implicitly_dependent_step(self, output_id: int, step_id: int):
                 )
 
     def _invoke_step(self, invocation_step: WorkflowInvocationStep) -> Optional[bool]:
+        assert invocation_step.workflow_step.module
         incomplete_or_none = invocation_step.workflow_step.module.execute(
             self.trans,
             self.progress,
@@ -735,6 +736,7 @@ def raw_to_galaxy(self, value: dict):
         return raw_to_galaxy(self.module_injector.trans.app, self.module_injector.trans.history, value)
 
     def _recover_mapping(self, step_invocation: WorkflowInvocationStep) -> None:
+        assert step_invocation.workflow_step.module
         try:
             step_invocation.workflow_step.module.recover_mapping(step_invocation, self)
         except modules.DelayedWorkflowEvaluation as de: