gxformat2 package#

Subpackages#

Submodules#

gxformat2.abstract module#

Module for exporting Galaxy workflows to CWL abstract interface.

gxformat2.abstract.from_dict(workflow_dict, subworkflow=False)[source]#

Convert Galaxy workflow into abstract CWL representation.

Accepts any workflow representation (raw dict, path, or typed model).

gxformat2.abstract.main(argv=None)[source]#: Entry point for script to export abstract interface.

gxformat2.converter module#

Functionality for converting a Format 2 workflow into a standard Galaxy workflow.

This module provides dict-returning wrapper functions used by Galaxy and Planemo. The typed API is gxformat2.normalized.to_native().

gxformat2.converter.main(argv=None)[source]#: Entry point for script to conversion from Format 2 interface.

gxformat2.converter.python_to_workflow(as_python, galaxy_interface=None, workflow_directory=None, import_options=None)[source]#: Convert a Format 2 workflow into standard Galaxy format from supplied dictionary.

gxformat2.converter.yaml_to_workflow(has_yaml, galaxy_interface=None, workflow_directory=None, import_options=None)[source]#: Convert a Format 2 workflow into standard Galaxy format from supplied stream.

gxformat2.cytoscape module#

Cytoscape.js visualization for Galaxy workflows.

class gxformat2.cytoscape.CytoscapeEdge(*, group: Literal['edges'] = 'edges', data: CytoscapeEdgeData)[source]#

Bases: BaseModel

A Cytoscape.js edge element.

data: CytoscapeEdgeData#

group: Literal['edges']#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gxformat2.cytoscape.CytoscapeEdgeData(*, id: str, source: str, target: str, input: str, output: str | None = None)[source]#

Bases: BaseModel

Data payload for a Cytoscape edge (step connection).

id: str#

input: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

output: str | None#

source: str#

target: str#

class gxformat2.cytoscape.CytoscapeElements(*, nodes: list[CytoscapeNode] = <factory>, edges: list[CytoscapeEdge] = <factory>, layout: CytoscapeLayout | None = None)[source]#

Bases: BaseModel

Complete set of Cytoscape.js elements for a workflow visualization.

edges: list[CytoscapeEdge]#

layout: CytoscapeLayout | None#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nodes: list[CytoscapeNode]#

to_dict() → dict[source]#

Serialize as {"elements": [...], "layout": {...}} wrapper.

Used by the CLI when --layout is non-default so the layout hint travels alongside the elements.

to_list() → list[dict][source]#: Serialize to the flat list-of-dicts format Cytoscape.js expects.

class gxformat2.cytoscape.CytoscapeLayout(*, name: str)[source]#

Bases: BaseModel

Layout hint emitted on non-default layouts.

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str#

class gxformat2.cytoscape.CytoscapeNode(*, group: Literal['nodes'] = 'nodes', data: CytoscapeNodeData, classes: list[str] = <factory>, position: CytoscapePosition | None = None)[source]#

Bases: BaseModel

A Cytoscape.js node element.

classes: list[str]#

data: CytoscapeNodeData#

group: Literal['nodes']#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

position: CytoscapePosition | None#

class gxformat2.cytoscape.CytoscapeNodeData(*, id: str, label: str, doc: str | None = None, tool_id: str | None = None, step_type: str = 'tool', repo_link: str | None = None)[source]#

Bases: BaseModel

Data payload for a Cytoscape node (input or step).

doc: str | None#

id: str#

label: str#

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

repo_link: str | None#

step_type: str#

tool_id: str | None#

class gxformat2.cytoscape.CytoscapePosition(*, x: int = 0, y: int = 0)[source]#

Bases: BaseModel

Cytoscape node position (x/y coordinates).

model_config = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

x: int#

y: int#

gxformat2.cytoscape.bakes_coordinates(layout: str) → bool[source]#

Layouts that bake coordinates into data.position.

All other layouts are hint-only and rely on the runtime renderer.

gxformat2.cytoscape.cytoscape_elements(workflow: dict[str, Any] | str | Path | GalaxyWorkflow | NormalizedFormat2, *, layout: str = 'preset') → CytoscapeElements[source]#

Build Cytoscape visualization elements from a Galaxy workflow.

Accepts anything normalized_format2() supports, plus an already normalized NormalizedFormat2 instance.

layout selects the placement strategy (default preset); see _layout.py and the cross-language spec for details.

gxformat2.cytoscape.is_layout_name(value: str) → bool[source]#: Return True if value is one of the supported Cytoscape layout names.

gxformat2.cytoscape.main(argv=None)[source]#: Entry point for building Cytoscape visualizations of Galaxy workflows.

gxformat2.cytoscape.render_html(elements: CytoscapeElements, layout: str = 'preset') → str[source]#

Return a standalone HTML page visualizing the workflow with Cytoscape.js.

The returned string is a complete HTML document suitable for writing to a file or embedding in a Jupyter notebook.

gxformat2.cytoscape.to_cytoscape(workflow_path: str, output_path=None, layout: str = 'preset')[source]#: Produce cytoscape output for supplied workflow path.

gxformat2.cytoscape.topological_positions(elements: CytoscapeElements) → dict[str, CytoscapePosition][source]#

Compute positions per the topological layering spec.

Returns a mapping keyed by node data.id.

gxformat2.export module#

Functionality for converting a standard Galaxy workflow into a format 2 workflow.

This module provides dict-returning wrapper functions used by Galaxy and Planemo. The typed API is gxformat2.normalized.to_format2().

gxformat2.export.from_galaxy_native(native_workflow_dict: dict[str, Any] | NativeGalaxyWorkflow, tool_interface=None, json_wrapper: bool = False, compact: bool = False, convert_tool_state: Callable[[dict], Dict[str, Any] | None] | None = None)[source]#

Convert native .ga workflow definition to a format2 workflow.

If convert_tool_state is provided it should be a callable accepting a native step dict and returning an optional dict representing the format2 state for that step. When the callable returns a dict, the step will carry state instead of tool_state; when it returns None the default tool_state passthrough is used.

gxformat2.export.main(argv=None)[source]#: Entry point for script to convert native workflows to Format 2.

gxformat2.interface module#

Compatibility shim for gxformat2.interface.

Deprecated since version 0.26.0: This module was removed in gxformat2 0.23 (PR #161) and has been restored solely as a compatibility shim for downstream consumers (notably Planemo). It may be removed again in a future release. New code should implement its own Galaxy importer interface or depend on Galaxy / BioBlend directly.

bioblend is no longer a hard dependency of gxformat2. Install the bioblend extra (pip install gxformat2[bioblend]) or install bioblend separately to use BioBlendImporterGalaxyInterface.

class gxformat2.interface.BioBlendImporterGalaxyInterface(**kwds)[source]#

Bases: ImporterGalaxyInterface

Implementation of ImporterGalaxyInterface using bioblend.

bioblend is imported lazily so importing gxformat2.interface does not require it to be installed.

import_workflow(workflow, **kwds)[source]#: Import Galaxy workflow using instance bioblend.GalaxyInstance object.

class gxformat2.interface.ImporterGalaxyInterface[source]#

Bases: object

An abstract interface describing Galaxy operations used by gxformat2.

Specifically containing definitions of operations required to load workflows into Galaxy.

abstractmethod import_workflow(workflow, **kwds)[source]#: Import a workflow via POST /api/workflows or comparable interface into Galaxy.

gxformat2.lint module#

Workflow linting entry point - main script.

gxformat2.lint.lint_best_practices(lint_context, nf2: NormalizedFormat2)[source]#: Lint best practices for a Format2 workflow (top-level + format2 step-level).

gxformat2.lint.lint_best_practices_format2(lint_context, workflow_dict)[source]#: Lint best practices for a Format2 Galaxy workflow.

gxformat2.lint.lint_best_practices_ga(lint_context, workflow_dict)[source]#

Lint best practices for a native Galaxy workflow.

Runs top-level checks on NormalizedFormat2 (for shared doc/creator/license interpretation) plus step-level checks on NormalizedNativeWorkflow so step messages reference native ids / labels / annotations rather than format2 sentinels like _unlabeled_step_1.

gxformat2.lint.lint_format2(lint_context, nf2, raw_dict: dict | None = None, path=None)[source]#

Lint a Format 2 Galaxy workflow and populate the corresponding LintContext.

Backward-compat: nf2 may be a raw dict (legacy Planemo signature), in which case it is expanded to NormalizedFormat2 internally. path is accepted for Planemo compatibility and currently ignored.

gxformat2.lint.lint_ga(lint_context, nnw, raw_dict: dict | None = None, path=None)[source]#

Lint a native Galaxy workflow and populate the corresponding LintContext.

Backward-compat: nnw may be a raw dict (legacy Planemo signature), in which case it is expanded to NormalizedNativeWorkflow internally. path is accepted for Planemo compatibility and currently ignored.

gxformat2.lint.lint_pydantic_validation(lint_context, workflow_dict, format2=False)[source]#

Validate workflow dict against pydantic schema models.

Tries strict model (extra=forbid) first. If strict fails, falls back to the lax model (extra=allow) to distinguish fundamental type errors from merely having extra/unknown fields.

gxformat2.lint.main(argv=None)[source]#: Script entry point for linting workflows.

gxformat2.linting module#

Generic utilities for linting.

Largely derived from galaxy.tool_util.lint. LintMessage is a str subclass so existing code that treats warn_messages / error_messages as lists of strings keeps working; new metadata (level, linter, json_pointer) is accessible as attributes.

class gxformat2.linting.LintContext(level='warn', training_topic=None, _pointer: str = '')[source]#

Bases: object

Track running status (state) of linting.

child(pointer_segment) → LintContext[source]#: Create child context whose default json_pointer is prefixed.

error(message: str, *args, linter: type | str | None = None, json_pointer: str | None = None, **kwds) → None[source]#: Track a linting error - a serious problem with the artifact preventing execution.

print_messages() → None[source]#: Print error messages and update state at the end of linting.

warn(message: str, *args, linter: type | str | None = None, json_pointer: str | None = None, **kwds) → None[source]#: Track a linting warning - a deviation from best practices.

class gxformat2.linting.LintLevel(*values)[source]#

Bases: str, Enum

Lint severity levels.

ALL = 'all'#

ERROR = 'error'#

WARN = 'warn'#

class gxformat2.linting.LintMessage(message: str, *, level: str = 'warn', linter: str | None = None, json_pointer: str = '')[source]#

Bases: str

A single lint emission: prose + structured metadata.

Subclassing str keeps "substring" in message and str(message) working for existing callers/tests.

json_pointer: str#

level: str#

linter: str | None#

class gxformat2.linting.Linter[source]#

Bases: object

Metadata-only base class for lint rules.

Subclasses carry class-level metadata; emission is performed by LintContext.warn / LintContext.error with linter=SubclassName.

applies_to: ClassVar[Tuple[str, ...]] = ()#

profile: ClassVar[str] = 'structural'#

severity: ClassVar[str] = 'warning'#

gxformat2.main module#

gxformat2.markdown_parse module#

Utilities for parsing “Galaxy Flavored Markdown”.

See markdown_util.py for loading objects and interacting with the rest of Galaxy. This file is meant to be relatively self contained and just used to “parse” and validate Galaxy Markdown. Keeping things isolated to allow re-use of these utilities in other projects (e.g. gxformat2).

gxformat2.markdown_parse.validate_galaxy_markdown(galaxy_markdown, internal=True)[source]#: Validate the supplied markdown and throw an ValueError with reason if invalid.

gxformat2.model module#

Abstractions for dealing with Format2 data.

gxformat2.model.ensure_step_position(step: dict, order_index: int)[source]#

Ensure step contains a position definition.

Modifies the input step dictionary.

gxformat2.model.get_native_step_type(gxformat2_step_dict: dict) → Literal['subworkflow', 'data_input', 'data_collection_input', 'tool', 'pause', 'pick_value', 'parameter_input'][source]#: Infer native galaxy step type from the gxformat2 step as a dict.

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

Parse a source reference into (step_label_or_id, output_name).

Deprecated: use gxformat2.normalized.resolve_source_reference directly.

gxformat2.model.setup_connected_values(value, key: str = '', append_to: dict[str, list] | None = None) → Any[source]#: Replace links with connected value.

gxformat2.model.steps_as_list(format2_workflow: dict, add_ids: bool = False, inputs_offset: int = 0, mutate: bool = False) → list[dict[str, Any]][source]#

Return steps as a list, converting ID map to list representation if needed.

This method does mutate the supplied steps, try to make progress toward not doing this.

Add keys as labels instead of IDs. Why am I doing this?

gxformat2.normalize module#

Convenience functions for accessing normalized workflow components.

steps(), inputs(), and outputs() return typed Pydantic model lists from any workflow representation. The _normalized variants return plain dicts for backward compatibility (used by Planemo).

gxformat2.normalize.inputs(workflow_dict: dict[str, Any] | str | PathLike | NormalizedFormat2 | NormalizedNativeWorkflow | GalaxyWorkflow | NativeGalaxyWorkflow | None = None, workflow_path: str | PathLike | None = None, options: ConversionOptions | None = None, expand: bool = False) → list[BaseInputParameter][source]#: Return normalized inputs as typed models.

gxformat2.normalize.inputs_normalized(workflow_dict: dict[str, Any] | str | PathLike | NormalizedFormat2 | NormalizedNativeWorkflow | GalaxyWorkflow | NativeGalaxyWorkflow | None = None, workflow_path: str | PathLike | None = None, options: ConversionOptions | None = None, expand: bool = False) → list[dict[str, Any]][source]#: Return normalized input steps as dicts.

Deprecated since version Use: inputs() for typed model access.

gxformat2.normalize.outputs(workflow_dict: dict[str, Any] | str | PathLike | NormalizedFormat2 | NormalizedNativeWorkflow | GalaxyWorkflow | NativeGalaxyWorkflow | None = None, workflow_path: str | PathLike | None = None, options: ConversionOptions | None = None, expand: bool = False) → list[WorkflowOutputParameter][source]#: Return normalized outputs as typed models.

gxformat2.normalize.outputs_normalized(workflow_dict: dict[str, Any] | str | PathLike | NormalizedFormat2 | NormalizedNativeWorkflow | GalaxyWorkflow | NativeGalaxyWorkflow | None = None, workflow_path: str | PathLike | None = None, options: ConversionOptions | None = None, expand: bool = False) → list[dict[str, Any]][source]#: Return normalized outputs as dicts.

Deprecated since version Use: outputs() for typed model access.

gxformat2.normalize.steps(workflow_dict: dict[str, Any] | str | PathLike | NormalizedFormat2 | NormalizedNativeWorkflow | GalaxyWorkflow | NativeGalaxyWorkflow | None = None, workflow_path: str | PathLike | None = None, options: ConversionOptions | None = None, expand: bool = False) → list[BaseInputParameter | NormalizedWorkflowStep][source]#: Return input parameters followed by steps as typed models.

Walk over a normalized step rep. across workflow formats.

Deprecated since version Use: steps() for typed model access.

Returns a list of dicts — input steps followed by tool/subworkflow steps.

gxformat2.options module#

Conversion options and URL resolution for workflow format conversion.

class gxformat2.options.ConversionOptions(workflow_directory: str | Path | None = None, encode_tool_state_json: bool = True, deduplicate_subworkflows: bool = False, state_encode_to_native: Callable[[dict, Dict[str, Any]], Dict[str, Any] | None] | None = None, state_encode_to_format2: Callable[[dict], Dict[str, Any] | None] | None = None, compact: bool = False, url_resolver: Callable[[str], dict[str, Any]] | None = None, strict_structure: bool = False)[source]#

Bases: object

Options for workflow format conversion and expansion.

Controls native↔Format2 conversion, subworkflow expansion, and URL resolution.

gxformat2.options.StateEncodeToFormat2Fn#

Callback to convert a native tool step’s tool_state to format2 state.

Accepts a native step dict (with tool_id, tool_version, tool_state). Returns a format2 state dict, or None to fall back to default tool_state passthrough.

alias of Callable[[dict], Dict[str, Any] | None] | None

gxformat2.options.StateEncodeToNativeFn#

Callback to encode format2 state back to native tool_state.

Accepts (step, state) where step is the partially-built native step dict and state is the format2 state dict after setup_connected_values processing. Returns {param_name: encoded_value} as clean dicts for native tool_state, or None to fall back to default dict passthrough (no JSON encoding).

alias of Callable[[dict, Dict[str, Any]], Dict[str, Any] | None] | None

gxformat2.options.UrlResolverFn#

Callback to fetch a URL and return a parsed workflow dict.

Accepts a URL string, returns a parsed dict (native or format2). Galaxy provides its own with allowlists/policy; gxformat2 provides a default via default_url_resolver().

alias of Callable[[str], dict[str, Any]] | None

gxformat2.options.default_url_resolver(url: str) → dict[str, Any][source]#

Fetch a URL and return a parsed workflow dict.

Handles: - base64:// URLs: base64-decode inline content - TRS URLs (GA4GH pattern): fetch descriptor endpoint, extract content - Plain URLs: HTTP GET, parse as YAML/JSON

gxformat2.options.is_trs_url(url: str) → bool[source]#: Check if a URL matches the GA4GH TRS v2 tools/versions pattern.

gxformat2.to_format2 module#

Convert a native Galaxy workflow to Format2.

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

gxformat2.to_native module#

Convert a Format2 workflow to native Galaxy format.

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

gxformat2.yaml module#

YAML loading utilities for gxformat2.

gxformat2.yaml.ordered_dump(data, stream=None, Dumper=<class 'yaml.dumper.SafeDumper'>, **kwds)[source]#: Safe and ordered dump of YAML to stream.

gxformat2.yaml.ordered_dump_to_path(as_dict: dict, path: str)[source]#: Safe and ordered dump of YAML to path.

gxformat2.yaml.ordered_load(stream, Loader=<class 'yaml.loader.SafeLoader'>, **kwds)[source]#: Safe and ordered load of YAML from stream.

gxformat2.yaml.ordered_load_path(path: str, **kwds)[source]#: Safe and ordered load of YAML from specified path.

Module contents#

The public interface or entry point for the Format 2 workflow code.

class gxformat2.ConversionOptions(workflow_directory: str | Path | None = None, encode_tool_state_json: bool = True, deduplicate_subworkflows: bool = False, state_encode_to_native: Callable[[dict, Dict[str, Any]], Dict[str, Any] | None] | None = None, state_encode_to_format2: Callable[[dict], Dict[str, Any] | None] | None = None, compact: bool = False, url_resolver: Callable[[str], dict[str, Any]] | None = None, strict_structure: bool = False)[source]

Bases: object

Options for workflow format conversion and expansion.

Controls native↔Format2 conversion, subworkflow expansion, and URL resolution.

class gxformat2.ImportOptions[source]: Bases: object

gxformat2.from_galaxy_native(native_workflow_dict: dict[str, Any] | NativeGalaxyWorkflow, tool_interface=None, json_wrapper: bool = False, compact: bool = False, convert_tool_state: Callable[[dict], Dict[str, Any] | None] | None = None)[source]

Convert native .ga workflow definition to a format2 workflow.

If convert_tool_state is provided it should be a callable accepting a native step dict and returning an optional dict representing the format2 state for that step. When the callable returns a dict, the step will carry state instead of tool_state; when it returns None the default tool_state passthrough is used.

gxformat2.python_to_workflow(as_python, galaxy_interface=None, workflow_directory=None, import_options=None)[source]: Convert a Format 2 workflow into standard Galaxy format from supplied dictionary.

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