gxformat2.normalized package#

Module contents#

Normalized and expanded workflow models with fully resolved types.

These models sit above the auto-generated pydantic schema models and provide uniform, narrowed types suitable for downstream consumers that want typed access without handling representational flexibility.

class gxformat2.normalized.ExpandedFormat2(*, class_: Literal['GalaxyWorkflow'] = 'GalaxyWorkflow', label: str | None = None, doc: str | None = None, inputs: list[~typing.Annotated[~gxformat2.schema.gxformat2.BaseInputParameter, ~pydantic.functional_serializers.SerializeAsAny()]]=<factory>, outputs: list[WorkflowOutputParameter] = <factory>, steps: list[ExpandedWorkflowStep] = <factory>, comments: list[TextComment | MarkdownComment | FrameComment | FreehandComment] = <factory>, report: Report | None = None, tags: list[str] = <factory>, creator: list[CreatorPerson | CreatorOrganization] | None = None, license: str | None = None, release: str | None = None, uuid: str | None = None)[source]#

Bases: NormalizedFormat2

Format2 workflow with all references expanded.

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

steps: list[ExpandedWorkflowStep]#

class gxformat2.normalized.ExpandedNativeStep(*, id: int, name: str | None = None, type: NativeStepType = NativeStepType.tool, label: str | None = None, annotation: str | None = None, when: str | None = None, content_id: str | None = None, content_source: str | None = None, tool_state: dict[str, ~typing.Any]=<factory>, tool_id: str | None = None, tool_version: str | None = None, tool_shed_repository: ToolShedRepository | None = None, tool_uuid: str | None = None, uuid: str | None = None, errors: str | None = None, position: StepPosition | None = None, input_connections: dict[str, list[~gxformat2.schema.native.NativeInputConnection]]=<factory>, inputs: list[NativeStepInput] = <factory>, outputs: list[NativeStepOutput] = <factory>, workflow_outputs: list[NativeWorkflowOutput] = <factory>, post_job_actions: dict[str, ~gxformat2.schema.native.NativePostJobAction]=<factory>, subworkflow: ExpandedNativeWorkflow | None = None, tool_representation: dict[str, ~typing.Any] | None=None, in_: dict[str, ~typing.Any] | None=None)[source]#

Bases: NormalizedNativeStep

Native step with subworkflow references resolved.

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

subworkflow: ExpandedNativeWorkflow | None#

class gxformat2.normalized.ExpandedNativeWorkflow(*, name: str | None = None, a_galaxy_workflow: str = 'true', format_version: str = '0.1', annotation: str | None = None, tags: list[str] = <factory>, version: int | None = None, license: str | None = None, release: str | None = None, uuid: str | None = None, creator: list[~typing.Annotated[~typing.Annotated[~gxformat2.schema.native.NativeCreatorPerson, ~pydantic.types.Tag(tag=NativeCreatorPerson)] | ~typing.Annotated[~gxformat2.schema.native.NativeCreatorOrganization, ~pydantic.types.Tag(tag=NativeCreatorOrganization)], ~pydantic.types.Discriminator(discriminator=~gxformat2.schema.native._discriminate_creator, custom_error_type=None, custom_error_message=None, custom_error_context=None)]] | None = None, report: ~gxformat2.schema.native.NativeReport | None = None, readme: str | None = None, help: str | None = None, logo_url: str | None = None, doi: list[str] | None = None, source_metadata: ~gxformat2.schema.native.NativeSourceMetadata | None = None, comments: list[~typing.Annotated[~typing.Annotated[~gxformat2.schema.native.NativeTextComment, ~pydantic.types.Tag(tag=NativeTextComment)] | ~typing.Annotated[~gxformat2.schema.native.NativeMarkdownComment, ~pydantic.types.Tag(tag=NativeMarkdownComment)] | ~typing.Annotated[~gxformat2.schema.native.NativeFrameComment, ~pydantic.types.Tag(tag=NativeFrameComment)] | ~typing.Annotated[~gxformat2.schema.native.NativeFreehandComment, ~pydantic.types.Tag(tag=NativeFreehandComment)], ~pydantic.types.Discriminator(discriminator=~gxformat2.schema.native._discriminate_comments, custom_error_type=None, custom_error_message=None, custom_error_context=None)]] = <factory>, steps: dict[str, ~gxformat2.normalized._conversion.ExpandedNativeStep] = <factory>, subworkflows: dict[str, ~gxformat2.normalized._native.NormalizedNativeWorkflow] | None = None)[source]#

Bases: NormalizedNativeWorkflow

Native workflow with all subworkflow references resolved.

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

steps: dict[str, ExpandedNativeStep]#

class gxformat2.normalized.ExpandedWorkflowStep(*, id: str, label: str | None = None, doc: str | None = None, type: WorkflowStepType = WorkflowStepType.tool, tool_id: str | None = None, tool_version: str | None = None, tool_shed_repository: ToolShedRepository | None = None, position: StepPosition | None = None, when: str | None = None, state: dict[str, ~typing.Any] | None=None, tool_state: dict[str, ~typing.Any] | None=None, runtime_inputs: list[str] | None = None, errors: str | None = None, uuid: str | None = None, in_: list[WorkflowStepInput] = <factory>, out: list[WorkflowStepOutput] = <factory>, post_job_actions: dict[str, ~gxformat2.schema.native.NativePostJobAction] | None=None, run: ExpandedFormat2 | GalaxyUserToolStub | None = None)[source]#

Bases: NormalizedWorkflowStep

Format2 step with run fully resolved.

Inline tool stubs (GalaxyUserToolStub) pass through expansion unchanged – they’re already self-contained and have no references to resolve.

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

run: ExpandedFormat2 | GalaxyUserToolStub | None#

class gxformat2.normalized.NormalizedFormat2(*, class_: Literal['GalaxyWorkflow'] = 'GalaxyWorkflow', label: str | None = None, doc: str | None = None, inputs: list[~typing.Annotated[~gxformat2.schema.gxformat2.BaseInputParameter, ~pydantic.functional_serializers.SerializeAsAny()]]=<factory>, outputs: list[WorkflowOutputParameter] = <factory>, steps: list[NormalizedWorkflowStep] = <factory>, comments: list[TextComment | MarkdownComment | FrameComment | FreehandComment] = <factory>, report: Report | None = None, tags: list[str] = <factory>, creator: list[CreatorPerson | CreatorOrganization] | None = None, license: str | None = None, release: str | None = None, uuid: str | None = None)[source]#

Bases: _DictMixin, BaseModel

A Format 2 Galaxy workflow with all union types resolved.

Steps, inputs, outputs, and comments are always lists. Input shorthands are expanded to full WorkflowInputParameter instances. Step ids are always populated.

class_: Literal['GalaxyWorkflow']#

comments: list[TextComment | MarkdownComment | FrameComment | FreehandComment]#

creator: list[CreatorPerson | CreatorOrganization] | None#

doc: str | None#

inputs: list[SerializeAsAny[BaseInputParameter]]#

property known_labels: set[str]#: All step and input labels/ids in this workflow.

label: str | None#

license: str | None#

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

outputs: list[WorkflowOutputParameter]#

release: str | None#

report: Report | None#

resolve_source(source: str) → SourceReference[source]#: Parse a source reference string (e.g. step/output) against this workflow’s labels.

steps: list[NormalizedWorkflowStep]#

tags: list[str]#

property unique_tools: frozenset[ToolReference]#: All unique (tool_id, tool_version) pairs in this workflow and its inline subworkflows.

uuid: str | None#

class gxformat2.normalized.NormalizedNativeStep(*, id: int, name: str | None = None, type: NativeStepType = NativeStepType.tool, label: str | None = None, annotation: str | None = None, when: str | None = None, content_id: str | None = None, content_source: str | None = None, tool_state: dict[str, ~typing.Any]=<factory>, tool_id: str | None = None, tool_version: str | None = None, tool_shed_repository: ToolShedRepository | None = None, tool_uuid: str | None = None, uuid: str | None = None, errors: str | None = None, position: StepPosition | None = None, input_connections: dict[str, list[~gxformat2.schema.native.NativeInputConnection]]=<factory>, inputs: list[NativeStepInput] = <factory>, outputs: list[NativeStepOutput] = <factory>, workflow_outputs: list[NativeWorkflowOutput] = <factory>, post_job_actions: dict[str, ~gxformat2.schema.native.NativePostJobAction]=<factory>, subworkflow: NormalizedNativeWorkflow | None = None, tool_representation: dict[str, ~typing.Any] | None=None, in_: dict[str, ~typing.Any] | None=None)[source]#

Bases: _DictMixin, BaseModel

A native workflow step with optional containers resolved to empty defaults.

tool_state is guaranteed to be a parsed dict.

annotation: str | None#

property connected_paths: frozenset[str]#: State paths that have incoming connections.

content_id: str | None#

content_source: str | None#

errors: str | None#

id: int#

in_: dict[str, Any] | None#

property inline_tool_class: str | None#: The class field of an inline tool representation, or None.

input_connections: dict[str, list[NativeInputConnection]]#

inputs: list[NativeStepInput]#

property is_inline_tool_step: bool#

The step carries an inline tool source via tool_representation.

Currently matches class: GalaxyUserTool (per-user dynamic tools); see INLINE_TOOL_CLASSES for the full list.

property is_input_step: bool#

property is_pause_step: bool#

property is_pick_value_step: bool#

property is_subworkflow_step: bool#

property is_tool_step: bool#

label: str | None#

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None#

outputs: list[NativeStepOutput]#

position: StepPosition | None#

post_job_actions: dict[str, NativePostJobAction]#

subworkflow: NormalizedNativeWorkflow | None#

tool_id: str | None#

tool_representation: dict[str, Any] | None#

tool_shed_repository: ToolShedRepository | None#

tool_state: dict[str, Any]#

tool_uuid: str | None#

tool_version: str | None#

type_: NativeStepType#

uuid: str | None#

when: str | None#

workflow_outputs: list[NativeWorkflowOutput]#

class gxformat2.normalized.NormalizedNativeWorkflow(*, name: str | None = None, a_galaxy_workflow: str = 'true', format_version: str = '0.1', annotation: str | None = None, tags: list[str] = <factory>, version: int | None = None, license: str | None = None, release: str | None = None, uuid: str | None = None, creator: list[~typing.Annotated[~typing.Annotated[~gxformat2.schema.native.NativeCreatorPerson, ~pydantic.types.Tag(tag=NativeCreatorPerson)] | ~typing.Annotated[~gxformat2.schema.native.NativeCreatorOrganization, ~pydantic.types.Tag(tag=NativeCreatorOrganization)], ~pydantic.types.Discriminator(discriminator=~gxformat2.schema.native._discriminate_creator, custom_error_type=None, custom_error_message=None, custom_error_context=None)]] | None = None, report: ~gxformat2.schema.native.NativeReport | None = None, readme: str | None = None, help: str | None = None, logo_url: str | None = None, doi: list[str] | None = None, source_metadata: ~gxformat2.schema.native.NativeSourceMetadata | None = None, comments: list[~typing.Annotated[~typing.Annotated[~gxformat2.schema.native.NativeTextComment, ~pydantic.types.Tag(tag=NativeTextComment)] | ~typing.Annotated[~gxformat2.schema.native.NativeMarkdownComment, ~pydantic.types.Tag(tag=NativeMarkdownComment)] | ~typing.Annotated[~gxformat2.schema.native.NativeFrameComment, ~pydantic.types.Tag(tag=NativeFrameComment)] | ~typing.Annotated[~gxformat2.schema.native.NativeFreehandComment, ~pydantic.types.Tag(tag=NativeFreehandComment)], ~pydantic.types.Discriminator(discriminator=~gxformat2.schema.native._discriminate_comments, custom_error_type=None, custom_error_message=None, custom_error_context=None)]] = <factory>, steps: dict[str, ~gxformat2.normalized._native.NormalizedNativeStep] = <factory>, subworkflows: dict[str, ~gxformat2.normalized._native.NormalizedNativeWorkflow] | None = None)[source]#

Bases: _DictMixin, BaseModel

A native Galaxy workflow with optional containers resolved to empty defaults.

Steps contain NormalizedNativeStep instances.

a_galaxy_workflow: str#

annotation: str | None#

comments: list[NativeComment]#

creator: list[NativeCreator] | None#

doi: list[str] | None#

format_version: str#

help: str | None#

license: str | None#

logo_url: str | None#

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None#

readme: str | None#

release: str | None#

report: NativeReport | None#

source_metadata: NativeSourceMetadata | None#

steps: dict[str, NormalizedNativeStep]#

subworkflows: dict[str, NormalizedNativeWorkflow] | None#

tags: list[str]#

property unique_tools: frozenset[ToolReference]#: All unique (tool_id, tool_version) pairs in this workflow and its inline subworkflows.

uuid: str | None#

version: int | None#

class gxformat2.normalized.NormalizedWorkflowStep(*, id: str, label: str | None = None, doc: str | None = None, type: WorkflowStepType = WorkflowStepType.tool, tool_id: str | None = None, tool_version: str | None = None, tool_shed_repository: ToolShedRepository | None = None, position: StepPosition | None = None, when: str | None = None, state: dict[str, ~typing.Any] | None=None, tool_state: dict[str, ~typing.Any] | None=None, runtime_inputs: list[str] | None = None, errors: str | None = None, uuid: str | None = None, in_: list[WorkflowStepInput] = <factory>, out: list[WorkflowStepOutput] = <factory>, post_job_actions: dict[str, ~gxformat2.schema.native.NativePostJobAction] | None=None, run: NormalizedFormat2 | GalaxyUserToolStub | ImportReference | str | None = None)[source]#

Bases: _DictMixin, BaseModel

A Format 2 workflow step with all union types resolved to canonical list form.

Ids are guaranteed populated.

property connected_paths: frozenset[str]#: State paths that have incoming connections.

doc: str | None#

errors: str | None#

id: str#

in_: list[WorkflowStepInput]#

property inline_tool_representation: dict[str, Any] | None#: The embedded tool source as a dict, or None if not an inline tool step.

property is_inline_tool_step: bool#

The step embeds an inline tool source via run.

True when run is a GalaxyUserToolStub or a dict whose class is in INLINE_TOOL_CLASSES. Native-validated stubs always normalize through GalaxyUserToolStub; the dict branch is defensive.

property is_pause_step: bool#

property is_pick_value_step: bool#

property is_subworkflow_step: bool#

property is_tool_step: bool#

label: str | None#

model_config = {'extra': 'ignore', 'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

out: list[WorkflowStepOutput]#

position: StepPosition | None#

post_job_actions: dict[str, NativePostJobAction] | None#

run: NormalizedFormat2 | GalaxyUserToolStub | ImportReference | str | None#

runtime_inputs: list[str] | None#

state: dict[str, Any] | None#

tool_id: str | None#

tool_shed_repository: ToolShedRepository | None#

tool_state: dict[str, Any] | None#

tool_version: str | None#

type_: WorkflowStepType#

uuid: str | None#

when: str | None#

class gxformat2.normalized.SourceReference(step_label: str, output_name: str)[source]#

Bases: NamedTuple

A parsed Format2 source reference (e.g. step_label/output_name).

output_name: str#: Alias for field number 1

step_label: str#: Alias for field number 0

class gxformat2.normalized.ToolReference(tool_id: str, tool_version: str | None)[source]#

Bases: NamedTuple

A unique tool identity within a workflow.

tool_id: str#: Alias for field number 0

tool_version: str | None#: Alias for field number 1

gxformat2.normalized.ensure_format2(workflow: GalaxyWorkflow | NormalizedFormat2, options: ConversionOptions, expand: Literal[True]) → ExpandedFormat2

gxformat2.normalized.ensure_format2(workflow: GalaxyWorkflow | NormalizedFormat2, options: ConversionOptions | None = None, expand: Literal[False] = False) → NormalizedFormat2

Ensure a workflow is returned as Format2 typed models.

Accepts native or Format2 inputs (raw dict/path or typed models), normalizing/converting as needed, and optionally expanding refs.

gxformat2.normalized.ensure_native(workflow: NativeGalaxyWorkflow | NormalizedNativeWorkflow, options: ConversionOptions, expand: Literal[True]) → ExpandedNativeWorkflow

gxformat2.normalized.ensure_native(workflow: NativeGalaxyWorkflow | NormalizedNativeWorkflow, options: ConversionOptions | None = None, expand: Literal[False] = False) → NormalizedNativeWorkflow

Ensure a workflow is returned as native typed models.

Accepts native or Format2 inputs (raw dict/path or typed models), normalizing/converting as needed, and optionally expanding refs.

Normalize and expand a Format2 workflow, resolving all references.

Resolves @import, URL, and TRS URL references in step run fields. Uses options.url_resolver (or the built-in default) for HTTP fetches.

Normalize and expand a native workflow, resolving all subworkflow references.

Resolves content_id URL references by fetching and converting them. Uses options.url_resolver (or the built-in default) for HTTP fetches.

gxformat2.normalized.native_input_to_format2_type(step: dict, tool_state: dict) → str | list[str][source]#: Return a Format2 input type (‘type’) from a native input step dictionary.

gxformat2.normalized.normalized_format2(workflow: dict[str, Any] | str | Path | PathLike[str] | GalaxyWorkflow, *, strict_structure: bool = False) → NormalizedFormat2[source]#

Normalize a Format 2 Galaxy workflow into a fully typed model.

Accepts a raw Format2 dict, a YAML/JSON file path, or a GalaxyWorkflow model. Raises ValueError if given a native Galaxy workflow — use ensure_format2() or to_format2() for cross-format input.

Handles $graph multi-workflow documents by extracting the main workflow and inlining #ref subworkflow references.

Returns a NormalizedFormat2 where steps/inputs/outputs are always lists, shorthands are expanded, and ids are populated.

When strict_structure is True, the raw input dict is validated against the strict schema (extra="forbid") before normalization, raising ValidationError on any unrecognised keys.

gxformat2.normalized.normalized_native(workflow: dict[str, Any] | str | Path | PathLike[str] | NativeGalaxyWorkflow, *, strict_structure: bool = False) → NormalizedNativeWorkflow[source]#

Normalize a native Galaxy workflow into a fully typed model.

Accepts a raw dict, a file path, or an already-parsed NativeGalaxyWorkflow. Returns a NormalizedNativeWorkflow where tool_state is always a dict, and all optional containers are resolved to empty defaults.

When strict_structure is True, the raw input dict is validated against the strict schema (extra="forbid") before normalization, raising ValidationError on any unrecognised keys.

gxformat2.normalized.resolve_source_reference(value: str, known_labels: set | dict) → SourceReference[source]#

Parse a source reference into (step_label_or_id, output_name).

Tries matching known labels first to handle labels containing ‘/’. Falls back to split on ‘/’ for numeric step IDs or unknown labels.

Convert a native Galaxy workflow to Format2.

Returns NormalizedFormat2 by default, or ExpandedFormat2 when expand=True.

Convert a Format2 workflow to native Galaxy format.

Returns NormalizedNativeWorkflow by default, or ExpandedNativeWorkflow when expand=True (resolving all URL/TRS/@import subworkflow references).