Fix mypy errors: enable pydantic plugin, fix type mismatches

Enable pydantic.mypy plugin to resolve alias kwargs (type_, in_,
format_version). Fix type narrowing with asserts, use concrete
comment types, proper model constructors (NativeStepInput,
NativeReport), typed POST_JOB_ACTIONS, creator conversion.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jmchilton

gxformat2/_labels.py

@@ -12,7 +12,7 @@ def __init__(self):
         self.seen_labels = set()
         self.anonymous_labels = 0
 
-    def ensure_new_output_label(self, label: str):
+    def ensure_new_output_label(self, label: str | None):
         """Ensure supplied label has value or generate an anonymous one."""
         if label is None:
             self.anonymous_labels += 1

gxformat2/normalized/_expanded.py

@@ -34,7 +34,7 @@ class ExpandedWorkflowStep(NormalizedWorkflowStep):
 class ExpandedFormat2(NormalizedFormat2):
     """Format2 workflow with all references expanded."""
 
-    steps: list[ExpandedWorkflowStep] = Field(default_factory=list)
+    steps: list[ExpandedWorkflowStep] = Field(default_factory=list)  # type: ignore[assignment]
 
 
 class ExpandedNativeStep(NormalizedNativeStep):
@@ -46,7 +46,7 @@ class ExpandedNativeStep(NormalizedNativeStep):
 class ExpandedNativeWorkflow(NormalizedNativeWorkflow):
     """Native workflow with all subworkflow references resolved."""
 
-    steps: dict[str, ExpandedNativeStep] = Field(default_factory=dict)
+    steps: dict[str, ExpandedNativeStep] = Field(default_factory=dict)  # type: ignore[assignment]
 
 
 ExpandedWorkflowStep.model_rebuild()
@@ -180,7 +180,7 @@ def _expand_native(wf: NormalizedNativeWorkflow, ctx: _ExpansionContext) -> Expa
         expanded_steps[key] = ExpandedNativeStep(**step_data, subworkflow=expanded_sub)
 
     # Expand subworkflows dict too
-    expanded_subworkflows: dict[str, ExpandedNativeWorkflow] | None = None
+    expanded_subworkflows: dict[str, NormalizedNativeWorkflow] | None = None
     if wf.subworkflows:
         expanded_subworkflows = {k: _expand_native(v, ctx) for k, v in wf.subworkflows.items()}
 

gxformat2/normalized/_format2.py

@@ -15,7 +15,6 @@
 from pydantic import BaseModel, ConfigDict, Field
 
 from gxformat2.schema.gxformat2 import (
-    BaseComment,
     CreatorOrganization,
     CreatorPerson,
     FrameComment,
@@ -126,6 +125,7 @@ def normalized_format2(
             workflow = from_galaxy_native(workflow)
         elif "$graph" in workflow and "class" not in workflow:
             workflow = _resolve_graph(workflow)
+        assert isinstance(workflow, dict)
         # Migrate legacy 'name' to 'label'
         if "name" in workflow and "label" not in workflow:
             workflow = {**workflow, "label": workflow["name"]}
@@ -137,6 +137,7 @@ def normalized_format2(
         if "steps" not in workflow:
             workflow = {**workflow, "steps": {}}
         workflow = GalaxyWorkflow.model_validate(workflow)
+    assert isinstance(workflow, GalaxyWorkflow)
     return _normalize_workflow(workflow)
 
 
@@ -309,6 +310,7 @@ def _normalize_step(step: WorkflowStep) -> NormalizedWorkflowStep:
         label=step.label,
         doc=_join_doc(step.doc),
         type_=step.type_,
+        in_=in_list,
         tool_id=step.tool_id,
         tool_version=step.tool_version,
         tool_shed_repository=step.tool_shed_repository,
@@ -319,7 +321,6 @@ def _normalize_step(step: WorkflowStep) -> NormalizedWorkflowStep:
         runtime_inputs=step.runtime_inputs,
         errors=step.errors,
         uuid=step.uuid,
-        in_=in_list,
         out=out_list,
         run=run,
     )
@@ -398,16 +399,16 @@ def _normalize_step_outputs(
 
 
 def _normalize_comments(
-    comments: list[BaseComment] | dict[str, BaseComment] | None,
+    comments: list[Format2Comment] | dict[str, Format2Comment] | None,
 ) -> list[Format2Comment]:
     if comments is None:
         return []
     if isinstance(comments, list):
-        return comments
+        return list(comments)
 
     result: list[Format2Comment] = []
     for key, comment in comments.items():
-        if isinstance(comment, BaseComment):
+        if isinstance(comment, (TextComment, MarkdownComment, FrameComment, FreehandComment)):
             if comment.label is None:
                 comment = comment.model_copy(update={"label": key})
             result.append(comment)
@@ -463,6 +464,7 @@ def _inline_graph_refs(workflow: dict[str, Any], lookup: dict[str, dict[str, Any
             continue
         run = step.get("run")
         if _is_graph_id_reference(run):
+            assert isinstance(run, str)
             ref_id = run[1:]
             if ref_id not in lookup:
                 raise Exception(f"$graph reference '{run}' not found in graph entries")

gxformat2/normalized/_native.py

@@ -135,6 +135,7 @@ def normalized_native(
             workflow = json.load(f)
     if isinstance(workflow, dict):
         workflow = load_native(workflow, strict=False)
+    assert isinstance(workflow, NativeGalaxyWorkflow)
     return _normalize_workflow(workflow)
 
 
@@ -164,7 +165,7 @@ def _normalize_workflow(wf: NativeGalaxyWorkflow) -> NormalizedNativeWorkflow:
         logo_url=wf.logo_url,
         doi=wf.doi,
         source_metadata=wf.source_metadata,
-        comments=[c.model_dump(by_alias=True) for c in wf.comments] if wf.comments else [],
+        comments=list(wf.comments) if wf.comments else [],
         steps=steps,
         subworkflows=subworkflows,
     )
@@ -187,6 +188,7 @@ def _normalize_step(step: NativeStep) -> NormalizedNativeStep:
         id=step.id or 0,
         name=step.name,
         type_=step.type_,
+        in_=step.in_,
         label=step.label,
         annotation=step.annotation,
         when=step.when,
@@ -207,5 +209,4 @@ def _normalize_step(step: NativeStep) -> NormalizedNativeStep:
         post_job_actions=step.post_job_actions or {},
         subworkflow=subworkflow,
         tool_representation=step.tool_representation,
-        in_=step.in_,
     )

gxformat2/to_format2.py

@@ -31,7 +31,6 @@
 )
 from .options import ConversionOptions
 from .schema.gxformat2 import (
-    BaseComment,
     CreatorOrganization,
     CreatorPerson,
     FrameComment,
@@ -361,8 +360,8 @@ def _build_pick_value_format2_step(
         label=display_label,
         doc=step.annotation or None,
         type_=WorkflowStepType.pick_value,
-        state=state,
         in_=in_list,
+        state=state,
         out=out_list,
         position=_convert_position(step.position) if not compact else None,
     )
@@ -465,7 +464,7 @@ def _to_source(output_name: str, label_map: dict[str, str], step_id: int) -> str
     return f"{output_label}/{output_name}"
 
 
-_CREATOR_CLASS_MAP: dict[str, type] = {
+_CREATOR_CLASS_MAP: dict[str, type[CreatorPerson] | type[CreatorOrganization]] = {
     "Person": CreatorPerson,
     "Organization": CreatorOrganization,
 }
@@ -484,7 +483,7 @@ def _convert_creators(
     return result
 
 
-_COMMENT_TYPE_MAP: dict[str, type[BaseComment]] = {
+_COMMENT_TYPE_MAP: dict[str, type[TextComment] | type[MarkdownComment] | type[FrameComment] | type[FreehandComment]] = {
     "text": TextComment,
     "markdown": MarkdownComment,
     "frame": FrameComment,

gxformat2/to_native.py

@@ -11,7 +11,7 @@
 import logging
 import uuid as uuid_mod
 from pathlib import Path
-from typing import Any, Literal, overload
+from typing import Any, Callable, Literal, overload, TypedDict
 
 from ._labels import Labels
 from .model import (
@@ -36,15 +36,22 @@
 )
 from .options import ConversionOptions
 from .schema.gxformat2 import (
-    BaseComment,
+    FrameComment,
+    FreehandComment,
     GalaxyWorkflow,
+    MarkdownComment,
+    TextComment,
     WorkflowInputParameter,
     WorkflowOutputParameter,
     WorkflowStepOutput,
 )
 from .schema.native import (
+    NativeCreatorOrganization,
+    NativeCreatorPerson,
     NativeInputConnection,
     NativePostJobAction,
+    NativeReport,
+    NativeStepInput,
     NativeStepType,
     NativeWorkflowOutput,
     StepPosition,
@@ -53,7 +60,13 @@
 
 log = logging.getLogger(__name__)
 
-POST_JOB_ACTIONS = {
+class _PJADef(TypedDict):
+    action_class: str
+    default: Any
+    arguments: Callable[..., dict[str, Any]]
+
+
+POST_JOB_ACTIONS: dict[str, _PJADef] = {
     "hide": {
         "action_class": "HideDatasetAction",
         "default": False,
@@ -180,7 +193,7 @@ def __init__(self, options: ConversionOptions):
 
     def step_id(self, label_or_id: str | int) -> int:
         if label_or_id in self.labels:
-            return self.labels[label_or_id]
+            return self.labels[label_or_id]  # type: ignore[index]
         return int(label_or_id)
 
     def step_output(self, value: str) -> tuple[int, str]:
@@ -212,6 +225,8 @@ def _build_native_workflow(
     # Convert comments
     comments = _build_native_comments(wf.comments, ctx)
 
+    native_creators = _convert_creators_to_native(wf.creator) if wf.creator else None
+    report = NativeReport(markdown=wf.report.markdown) if wf.report else None
     return NormalizedNativeWorkflow(
         a_galaxy_workflow="true",
         format_version="0.1",
@@ -221,8 +236,8 @@ def _build_native_workflow(
         tags=wf.tags,
         license=wf.license,
         release=wf.release,
-        creator=wf.creator,
-        report={"markdown": wf.report.markdown} if wf.report else None,
+        creator=native_creators,
+        report=report,
         steps=native_steps,
         comments=comments,
     )
@@ -244,7 +259,7 @@ def _build_input_step(
     else:
         multiple = False
 
-    type_str = input_type.value if hasattr(input_type, "value") else str(input_type) if input_type else "data"
+    type_str = input_type.value if input_type is not None and hasattr(input_type, "value") else str(input_type) if input_type else "data"
     if type_str in ("File", "data", "data_input"):
         step_type = NativeStepType.data_input
     elif type_str in ("collection", "data_collection", "data_collection_input"):
@@ -290,13 +305,13 @@ def _build_input_step(
     return NormalizedNativeStep(
         id=order_index,
         type_=step_type,
+        in_=in_,
         label=label,
         name=raw_label,
         annotation=_join_doc(inp.doc) or "",
         tool_state=tool_state,
         position=position,
-        inputs=[{"name": raw_label, "description": ""}],
-        in_=in_,
+        inputs=[NativeStepInput(name=raw_label, description="")],
     )
 
 
@@ -482,7 +497,7 @@ def _build_pause_step(
         annotation=step.doc or "",
         tool_state={"name": name},
         input_connections=input_connections,
-        inputs=[{"name": name, "description": ""}],
+        inputs=[NativeStepInput(name=name, description="")],
         position=position,
         uuid=step.uuid,
     )
@@ -516,7 +531,7 @@ def _build_pick_value_step(
         tool_state=tool_state,
         input_connections=input_connections,
         post_job_actions=post_job_actions,
-        inputs=[{"name": name, "description": ""}],
+        inputs=[NativeStepInput(name=name, description="")],
         position=position,
         uuid=step.uuid,
     )
@@ -636,7 +651,7 @@ def _wire_workflow_outputs(
 
 
 def _build_native_comments(
-    comments: list[BaseComment],
+    comments: list[TextComment | MarkdownComment | FrameComment | FreehandComment],
     ctx: _ConversionContext,
 ) -> list:
     """Convert Format2 comments to native format."""
@@ -726,6 +741,23 @@ def _default_position(position: Any, order_index: int) -> StepPosition:
     return StepPosition(left=10 * order_index, top=10 * order_index)
 
 
+_NATIVE_CREATOR_MAP: dict[str, type[NativeCreatorPerson] | type[NativeCreatorOrganization]] = {
+    "Person": NativeCreatorPerson,
+    "Organization": NativeCreatorOrganization,
+}
+
+
+def _convert_creators_to_native(
+    creators: list,
+) -> list[NativeCreatorPerson | NativeCreatorOrganization]:
+    result: list[NativeCreatorPerson | NativeCreatorOrganization] = []
+    for c in creators:
+        d = c.model_dump(by_alias=True, exclude_none=True) if hasattr(c, "model_dump") else c
+        cls = _NATIVE_CREATOR_MAP.get(d.get("class", ""), NativeCreatorPerson)
+        result.append(cls.model_validate(d))
+    return result
+
+
 def _join_doc(doc: str | list[str] | None) -> str | None:
     if doc is None:
         return None

setup.cfg

@@ -1,5 +1,9 @@
 [mypy]
 ignore_missing_imports = true
+plugins = pydantic.mypy
+
+[pydantic-mypy]
+init_typed = true
 
 [nosetests]
 verbosity=1