# Auto-generated by schema-salad-plus-pydantic — do not edit.
from __future__ import annotations
import json
from enum import Enum
from pathlib import Path
from typing import Any, Annotated, Literal, Union
from pydantic import BaseModel, ConfigDict, Field, Discriminator, Tag
[docs]
def parser_info() -> str:
return ""
[docs]
class PrimitiveType(str, Enum):
"""Salad data types are based on Avro schema declarations. Refer to the
[Avro schema declaration documentation](https://avro.apache.org/docs/current/spec.html#schemas) for
detailed information.
null: no value
boolean: a binary value
int: 32-bit signed integer
long: 64-bit signed integer
float: single precision (32-bit) IEEE 754 floating-point number
double: double precision (64-bit) IEEE 754 floating-point number
string: Unicode character sequence"""
null = "null"
boolean = "boolean"
int = "int"
long = "long"
float = "float"
double = "double"
string = "string"
[docs]
class NativeStepType(str, Enum):
"""Step module types in the native Galaxy workflow format.
data_input: A single dataset input to the workflow.
data_collection_input: A dataset collection input to the workflow.
parameter_input: A typed parameter input (text, integer, float, boolean, color).
tool: A Galaxy tool execution step.
subworkflow: An embedded or referenced sub-workflow.
pause: A manual pause point that halts execution until user intervention.
pick_value: Select the first non-null value from multiple inputs."""
data_input = "data_input"
data_collection_input = "data_collection_input"
parameter_input = "parameter_input"
tool = "tool"
subworkflow = "subworkflow"
pause = "pause"
pick_value = "pick_value"
def _discriminate_creator(v: Any) -> str:
disc_map: dict[str, str] = {"Person": "NativeCreatorPerson", "Organization": "NativeCreatorOrganization"}
if isinstance(v, dict):
disc_val: str = str(v.get("class", ""))
else:
disc_val = str(getattr(v, "class_", ""))
return disc_map.get(disc_val, disc_val)
def _discriminate_comments(v: Any) -> str:
disc_map: dict[str, str] = {"text": "NativeTextComment", "markdown": "NativeMarkdownComment", "frame": "NativeFrameComment", "freehand": "NativeFreehandComment"}
if isinstance(v, dict):
disc_val: str = str(v.get("type", ""))
else:
disc_val = str(getattr(v, "type_", ""))
return disc_map.get(disc_val, disc_val)
[docs]
class Documented(BaseModel):
model_config = ConfigDict(populate_by_name=True, extra="forbid")
doc: None | str | list[str] = Field(default=None, description="A documentation string for this object, or an array of strings which should be concatenated.")
[docs]
class RecordField(Documented):
"""A field of a record."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: str = Field(description="The name of the field")
type_: PrimitiveType | RecordSchema | EnumSchema | ArraySchema | str | list[PrimitiveType | RecordSchema | EnumSchema | ArraySchema | str] = Field(alias="type", description="The field type")
[docs]
class RecordSchema(BaseModel):
model_config = ConfigDict(populate_by_name=True, extra="forbid")
fields: None | list[RecordField] = Field(default=None, description="Defines the fields of the record.")
type_: Literal["record"] = Field(default="record", alias="type", description="Must be `record`")
[docs]
class EnumSchema(BaseModel):
"""Define an enumerated type."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
symbols: list[str] = Field(description="Defines the set of valid symbols.")
type_: Literal["enum"] = Field(default="enum", alias="type", description="Must be `enum`")
[docs]
class ArraySchema(BaseModel):
model_config = ConfigDict(populate_by_name=True, extra="forbid")
items: PrimitiveType | RecordSchema | EnumSchema | ArraySchema | str | list[PrimitiveType | RecordSchema | EnumSchema | ArraySchema | str] = Field(description="Defines the type of the array elements.")
type_: Literal["array"] = Field(default="array", alias="type", description="Must be `array`")
[docs]
class HasUUID(BaseModel):
model_config = ConfigDict(populate_by_name=True, extra="forbid")
uuid: None | str = Field(default=None, description="UUID uniquely representing this element.")
[docs]
class HasStepErrors(BaseModel):
model_config = ConfigDict(populate_by_name=True, extra="forbid")
errors: None | str = Field(default=None, description="During Galaxy export there may be some problem validating the tool state, tool used, etc.. that will be indicated by this field. The Galaxy user should be warned of these problems before the workfl...")
[docs]
class HasStepPosition(BaseModel):
model_config = ConfigDict(populate_by_name=True, extra="forbid")
position: None | StepPosition = Field(default=None)
[docs]
class StepPosition(BaseModel):
"""This field specifies the location of the step's node when rendered in the workflow editor."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
top: float | int = Field(description="Relative vertical position of the step's node when rendered in the workflow editor.")
left: float | int = Field(description="Relative horizontal position of the step's node when rendered in the workflow editor.")
[docs]
class SampleSheetColumnDefinition(BaseModel):
"""Describes one column of a sample-sheet collection input.
Used in `column_definitions` on a `collection_type: sample_sheet[:<type>]`
workflow input."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: str = Field(description="Column name. Must not contain special characters (matches `^[\\w\\-_ \\?]*$`).")
description: None | str = Field(default=None, description="Optional human-readable column description.")
type_: Literal["string", "int", "float", "boolean", "element_identifier"] = Field(default="string", alias="type", description="Value type for this column. One of `string`, `int`, `float`, `boolean`, or `element_identifier`. Mirrors Galaxy's runtime `SampleSheetColumnType`.")
optional: bool = Field(description="If true, rows may omit a value for this column.")
default_value: None | str | int | float | bool = Field(default=None, description="Default value used when a row omits this column. Type must be compatible with `type` - validated by the pydantic post-validator.")
validators: None | list[Any] = Field(default=None, description="Galaxy-style parameter validators. Modelled as opaque records here - full validator schema lives in galaxy.tool_util_models.")
restrictions: None | list[str | int | float | bool] = Field(default=None, description="Closed set of permitted values for this column. Item type must be compatible with the column `type` (post-validated).")
suggestions: None | list[str | int | float | bool] = Field(default=None, description="Open suggestion list for this column.")
[docs]
class RecordFieldDefinition(BaseModel):
"""Describes one field of a `record` collection input.
Used in `fields` on a `collection_type` containing `record` (e.g.
`record`, `list:record`, `sample_sheet:record`). Mirrors a subset of
the CWL `InputRecordSchema` shape that Galaxy persists on
`DatasetCollection.fields`."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: str = Field(description="Field name. Must equal the corresponding element identifier in the materialized record collection.")
type_: Literal["File", "null", "boolean", "int", "float", "string"] | list[Literal["File", "null", "boolean", "int", "float", "string"]] = Field(default="File", alias="type", description="Field value type. A subset of the CWL primitive types: `File`, `null`, `boolean`, `int`, `float`, `string`. May be a list to express a union (e.g. `[\"File\", \"null\"]` for an optional file).")
format: None | str = Field(default=None, description="Optional Galaxy datatype hint for `File`-typed fields.")
[docs]
class WorkflowTextOption(BaseModel):
"""A `{value, label}` option used in `restrictions` or `suggestions` on a
text workflow parameter. Plain strings are also accepted in those
arrays as shorthand for `{value: <str>, label: <str>}`."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
value: str = Field(description="Machine value submitted to the connected tool input.")
label: None | str = Field(default=None, description="Human label shown in Galaxy. Defaults to `value` when omitted.")
[docs]
class NativeStepOutput(BaseModel):
"""Declares an output produced by a step, with its name and datatype."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: str = Field(description="Name of the output.")
type_: None | str = Field(default=None, alias="type", description="Galaxy datatype extension for the output (e.g. ``bam``, ``txt``, ``tabular``). The value ``input`` means the type is determined at runtime from the input data.")
[docs]
class NativeWorkflowOutput(BaseModel):
"""Designates a step output as a workflow-level output. These appear in the
``workflow_outputs`` array on each step."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
label: None | str = Field(default=None, description="User-visible label for this workflow output. Must be unique across the workflow.")
output_name: str = Field(description="Name of the step output to expose as a workflow output.")
uuid: None | str = Field(default=None, description="UUID for this workflow output.")
[docs]
class NativePostJobAction(BaseModel):
"""An action applied to a step's output after tool execution. These objects
appear as values in the ``post_job_actions`` dictionary, keyed by
compound strings of the form ``{ActionType}{OutputName}``
(e.g. ``HideDatasetActionout_pairs``).
Common action types: ``HideDatasetAction``, ``RenameDatasetAction``,
``DeleteIntermediatesAction``, ``ChangeDatatypeAction``,
``TagDatasetAction``, ``ColumnSetAction``."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
action_type: str = Field(description="The action type identifier (e.g. ``HideDatasetAction``).")
output_name: str | None = Field(default=None, description="The step output this action applies to. Optional: action types that operate on the step as a whole (e.g. ``ValidateOutputsAction``) omit this field.")
action_arguments: dict[str, Any] | None = Field(default=None, description="Action-specific arguments. For ``RenameDatasetAction``: ``{\"newname\": \"...\"}``; for ``ChangeDatatypeAction``: ``{\"newtype\": \"tabular\"}``; for ``TagDatasetAction``: ``{\"tags\": \"name:tag\"...")
[docs]
class NativeStep(HasStepErrors, HasStepPosition, HasUUID, ReferencesTool):
"""A step in a native Galaxy workflow. Steps are keyed by string integers
(``"0"``, ``"1"``, ...) in the ``steps`` dictionary. The key serves as
an external ID for connection wiring.
The ``type`` field determines the step's behavior. All step types share
common fields; type-specific fields are optional and only relevant for
their respective type.
# Tool State
The ``tool_state`` field contains the tool's parameter configuration.
Traditionally this is a JSON-encoded string (double-encoded: each top-level
value is itself a JSON string), but it may also be a plain dictionary when
unencoded tool state is used. Connected parameters are represented as
``{"__class__": "ConnectedValue"}``, runtime parameters as
``{"__class__": "RuntimeValue"}``."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: None | str = Field(default=None, description="Display name. For tool steps this is the tool's name; for input steps it is the generic type name (e.g. \"Input dataset\").")
id: None | int = Field(default=None, description="Step ID. Matches the dictionary key in the ``steps`` object.")
type_: None | NativeStepType = Field(default=None, alias="type", description="Step module type.")
label: None | str = Field(default=None, description="User-friendly step label. Must be unique within the workflow.")
annotation: None | str = Field(default=None, description="Human-readable step description.")
when: None | str = Field(default=None, description="Conditional execution expression. Typically ``$(inputs.when)`` paired with a boolean input connection on the ``when`` key.")
content_id: None | str = Field(default=None, description="Tool ID or content identifier. Same as ``tool_id`` for tool steps; may reference a subworkflow key for subworkflow steps.")
tool_state: str | dict[str, Any] | None = Field(default=None, description="Tool parameter configuration. Traditionally a JSON-encoded string (double-encoded: each top-level value is itself a JSON string), but may also be a plain dict when ``encode_tool_state`` is disabled...")
tool_uuid: None | str = Field(default=None, description="UUID for dynamic (inline-defined) tools.")
input_connections: dict[str, NativeInputConnection | list[NativeInputConnection]] | None = Field(default=None, description="Connection wiring from other steps' outputs to this step's inputs. Keys are input parameter names (using ``|`` for nested parameters). Values are a NativeInputConnection object or an array of them ...")
inputs: None | list[NativeStepInput] = Field(default=None, description="Input parameter metadata (descriptions, not connections).")
outputs: None | list[NativeStepOutput] = Field(default=None, description="Output declarations with names and datatypes.")
workflow_outputs: None | list[NativeWorkflowOutput] = Field(default=None, description="Outputs designated as workflow-level outputs. Steps with an empty array have all their outputs hidden by default.")
post_job_actions: dict[str, NativePostJobAction] | None = Field(default=None, description="Post-job actions applied to outputs after tool execution. Keys are compound strings of the form ``{ActionType}{OutputName}``. Values are NativePostJobAction objects.")
subworkflow: NativeGalaxyWorkflow | None = Field(default=None, description="Embedded subworkflow definition. A complete native workflow document (with ``a_galaxy_workflow``, ``format-version``, ``steps``, etc.) nested inside this step.")
tool_representation: dict[str, Any] | None = Field(default=None, description="Embedded tool definition for user-defined (dynamic) tools. Present when the step uses a ``GalaxyUserTool`` instead of a registered tool. Contains the full tool definition including ``class: GalaxyU...")
in_: dict[str, Any] | None = Field(default=None, alias="in", description="Default values for step inputs. Keys are parameter names, values are objects with a ``default`` key.")
[docs]
class NativeReport(BaseModel):
"""Workflow invocation report template."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
markdown: str = Field(description="Galaxy-flavored Markdown content for the invocation report. Supports template directives like ``history_dataset_as_image(output=\"...\")``.")
[docs]
class BaseNativeCreator(BaseModel):
"""Base fields shared by all creator types, corresponding to schema.org
Thing properties common to both Person and Organization."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: None | str = Field(default=None, description="Full name of the person or organization.")
identifier: None | str = Field(default=None, description="Persistent identifier, typically an ORCID URL (e.g. ``https://orcid.org/0000-0001-2345-6789``) or bare ORCID.")
url: None | str = Field(default=None, description="Website or profile URL.")
email: None | str = Field(default=None, description="Email address. May include a ``mailto:`` prefix.")
image: None | str = Field(default=None, description="URL to an image or avatar.")
address: None | str = Field(default=None, description="Physical or mailing address.")
alternateName: None | str = Field(default=None, description="An alternate name or alias.")
telephone: None | str = Field(default=None, description="Telephone number.")
faxNumber: None | str = Field(default=None, description="Fax number.")
[docs]
class NativeCreatorPerson(BaseNativeCreator):
"""A person who created or contributed to the workflow.
Corresponds to a `schema.org Person <https://schema.org/Person>`_."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class_: Literal["Person"] = Field(default="Person", alias="class", description="Creator type discriminator (``Person``).")
givenName: None | str = Field(default=None, description="Given (first) name.")
familyName: None | str = Field(default=None, description="Family (last) name.")
honorificPrefix: None | str = Field(default=None, description="Honorific prefix (e.g. ``Dr``, ``Prof``).")
honorificSuffix: None | str = Field(default=None, description="Honorific suffix (e.g. ``M.D.``, ``PhD``).")
jobTitle: None | str = Field(default=None, description="Job title or role.")
[docs]
class NativeCreatorOrganization(BaseNativeCreator):
"""An organization that created or contributed to the workflow.
Corresponds to a `schema.org Organization <https://schema.org/Organization>`_."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
class_: Literal["Organization"] = Field(default="Organization", alias="class", description="Creator type discriminator (``Organization``).")
[docs]
class NativeGalaxyWorkflow(HasUUID):
"""A Galaxy native workflow document (.ga format). This is the canonical
serialization format used internally by Galaxy.
The native format mirrors the database schema directly -- steps are keyed
by numeric IDs, connections reference step IDs, and tool state is stored
as double-encoded JSON strings.
The format version has been ``"0.1"`` since inception. New features are
added without version bumps."""
model_config = ConfigDict(populate_by_name=True, extra="forbid")
name: None | str = Field(default=None, description="Workflow display name.")
class_: Literal["NativeGalaxyWorkflow"] = Field(default="NativeGalaxyWorkflow", alias="class")
a_galaxy_workflow: Literal['true'] = Field(description="Format marker. Always the string ``\"true\"``.")
format_version: Literal['0.1'] = Field(alias="format-version", description="Format version. Always ``\"0.1\"`` currently. In the JSON document this field is written as ``format-version``.")
annotation: None | str = Field(default=None, description="Human-readable workflow description.")
tags: None | list[str] = Field(default=None, description="Classification tags for the workflow.")
version: None | int = Field(default=None, description="Internal revision counter.")
license: None | str = Field(default=None, description="SPDX license identifier (e.g. ``\"MIT\"``, ``\"CC-BY-4.0\"``).")
release: None | str = Field(default=None, description="Semantic version for published workflows.")
# Discriminated union on 'class'
creator: list[Annotated[Annotated[NativeCreatorPerson, Tag("NativeCreatorPerson")] | Annotated[NativeCreatorOrganization, Tag("NativeCreatorOrganization")], Discriminator(_discriminate_creator)]] | None = None
report: None | NativeReport = Field(default=None, description="Workflow invocation report template.")
readme: None | str = Field(default=None, description="Detailed workflow description in Markdown.")
help: None | str = Field(default=None, description="Help text in Markdown.")
logo_url: None | str = Field(default=None, description="URL to workflow logo image.")
doi: None | list[str] = Field(default=None, description="Associated DOIs.")
source_metadata: None | NativeSourceMetadata = Field(default=None, description="Provenance tracking metadata for workflows imported from TRS or URLs.")
# Discriminated union on 'type'
comments: list[Annotated[Annotated[NativeTextComment, Tag("NativeTextComment")] | Annotated[NativeMarkdownComment, Tag("NativeMarkdownComment")] | Annotated[NativeFrameComment, Tag("NativeFrameComment")] | Annotated[NativeFreehandComment, Tag("NativeFreehandComment")], Discriminator(_discriminate_comments)]] | None = None
steps: dict[str, NativeStep] | None = Field(default=None, description="Step dictionary keyed by string integers (``\"0\"``, ``\"1\"``, ...). Values are NativeStep objects. Steps are processed in sorted numeric order during import.")
subworkflows: dict[str, NativeGalaxyWorkflow] | None = Field(default=None, description="Map of locally-defined subworkflow dicts, used during format conversion.")
# Rebuild models for forward references
Documented.model_rebuild()
RecordField.model_rebuild()
RecordSchema.model_rebuild()
EnumSchema.model_rebuild()
ArraySchema.model_rebuild()
HasUUID.model_rebuild()
HasStepErrors.model_rebuild()
HasStepPosition.model_rebuild()
StepPosition.model_rebuild()
ReferencesTool.model_rebuild()
SampleSheetColumnDefinition.model_rebuild()
RecordFieldDefinition.model_rebuild()
WorkflowTextOption.model_rebuild()
ToolShedRepository.model_rebuild()
NativeStepInput.model_rebuild()
NativeStepOutput.model_rebuild()
NativeWorkflowOutput.model_rebuild()
NativeInputConnection.model_rebuild()
NativePostJobAction.model_rebuild()
NativeTextCommentData.model_rebuild()
NativeMarkdownCommentData.model_rebuild()
NativeFrameCommentData.model_rebuild()
NativeFreehandCommentData.model_rebuild()
BaseNativeComment.model_rebuild()
NativeTextComment.model_rebuild()
NativeMarkdownComment.model_rebuild()
NativeFrameComment.model_rebuild()
NativeFreehandComment.model_rebuild()
NativeStep.model_rebuild()
NativeReport.model_rebuild()
BaseNativeCreator.model_rebuild()
NativeCreatorPerson.model_rebuild()
NativeCreatorOrganization.model_rebuild()
NativeSourceMetadata.model_rebuild()
NativeGalaxyWorkflow.model_rebuild()
[docs]
def load_document(path: str | Path) -> NativeGalaxyWorkflow | list[NativeGalaxyWorkflow]:
"""Load and validate a document from a JSON file."""
with open(path) as f:
data = json.load(f)
if isinstance(data, list):
return [_load_single(item) for item in data]
return _load_single(data)
def _load_single(data: dict[str, Any]) -> NativeGalaxyWorkflow:
"""Load a single document dict."""
return NativeGalaxyWorkflow.model_validate(data)