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.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.lint_profiles module#

Lint profile catalog loader.

Profiles group lint-rule IDs into named sets (structural, best-practices, release). Unknown IDs are tolerated — audit tooling compares against the registered Linter subclasses to flag unimplemented entries as INFO.

class gxformat2.lint_profiles.LintProfile(*, id: str, description: str = '', rules: list[str])[source]#

Bases: BaseModel

Named set of lint-rule IDs loaded from lint_profiles.yml.

description: str#

id: str#

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

rules: list[str]#

gxformat2.lint_profiles.iwc_rule_ids() → set[str][source]#: Union of structural, best-practices, and release rule IDs.

gxformat2.lint_profiles.lint_profiles_by_id() → dict[str, LintProfile][source]#: Return profiles keyed by id.

gxformat2.lint_profiles.load_lint_profiles() → list[LintProfile][source]#: Parse lint_profiles.yml into validated LintProfile models.

gxformat2.lint_profiles.rules_for_profile(profile_id: str) → list[str][source]#: Return the ordered rule ID list for a named profile.

gxformat2.lint_rules module#

Lint rule classes.

Each Linter subclass carries metadata only (severity, applies_to, profile). Emission happens in gxformat2/lint.py via LintContext.warn / LintContext.error with linter=<Subclass>.

class gxformat2.lint_rules.NativeStepKeyNotInteger[source]#

Bases: Linter

Native workflow step keys must be string representations of integers.

applies_to: ClassVar[tuple[str, ...]] = ('native',)#

profile: ClassVar[str] = 'structural'#

severity: ClassVar[str] = 'error'#

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.

error_messages: list[LintMessage]#

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.

warn_messages: list[LintMessage]#

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.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.schema_rules module#

Schema-rule catalog loader.

Schema rules describe checks already enforced by the pydantic decode layer. Each rule ships with positive and negative fixtures; the catalog contract is tested end-to-end by running fixtures through validators — not by inspecting ValidationError shapes.

class gxformat2.schema_rules.AppliesTo(*values)[source]#

Bases: str, Enum

Which workflow format a schema rule applies to.

format2 = 'format2'#

native = 'native'#

class gxformat2.schema_rules.SchemaRule(*, id: str, severity: Severity, applies_to: list[AppliesTo], scope: Scope, description: str = '', tests: SchemaRuleTests)[source]#

Bases: BaseModel

Declarative schema-rule entry loaded from schema_rules.yml.

applies_to: list[AppliesTo]#

description: str#

id: str#

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

scope: Scope#

severity: Severity#

tests: SchemaRuleTests#

class gxformat2.schema_rules.SchemaRuleTests(*, positive: list[str], negative: list[str])[source]#

Bases: BaseModel

Positive/negative fixture references for a schema rule.

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

negative: list[str]#

positive: list[str]#

class gxformat2.schema_rules.Scope(*values)[source]#

Bases: str, Enum

Validator flavors that reject the negative fixture.

both: both lax and strict reject (e.g. missing required field).
strict: only strict rejects (e.g. unknown extra field).
lax: only lax rejects — unusual, reserved for completeness.

both = 'both'#

lax = 'lax'#

strict = 'strict'#

class gxformat2.schema_rules.Severity(*values)[source]#

Bases: str, Enum

Rule severity level.

error = 'error'#

warning = 'warning'#

gxformat2.schema_rules.load_schema_rules() → list[SchemaRule][source]#: Parse schema_rules.yml into validated SchemaRule models.

gxformat2.testing module#

Reusable declarative test harness for YAML-driven workflow operation tests.

Extracts the test infrastructure from gxformat2’s own declarative tests into an importable module. Callers inject their own operations dict and fixture loader, making this usable from any project (Galaxy, Planemo, etc.).

Path element types for assertions:

str: dict key lookup (falls back to attribute access)
int: list index
“$length”: terminal, returns len(current object)
{field: value}: find first list item where item.field == value

Assertion modes:

value: exact equality
value_contains: substring containment
value_any_contains: any element in a list contains substring
value_set: unordered set comparison
value_matches: regex match
value_truthy / value_falsy: boolean-ish checks
value_type: isinstance check (“dict”, “list”, “str”, “int”, “float”, “bool”)
value_absent: assert that path does NOT resolve (key/attribute missing)

Bases: BaseModel

A single path-based assertion against an operation result.

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

path: list[str | int | dict[str, Any]]#

value: Any#

value_absent: bool | None#

value_any_contains: str | None#

value_contains: str | None#

value_falsy: bool | None#

value_matches: str | None#

value_set: list[Any] | None#

value_truthy: bool | None#

value_type: str | None#

class gxformat2.testing.DeclarativeTestSuite(operations: dict[str, Callable[[...], Any]], load_fixture: Callable[[str], Any], expectations_dir: str | None = None, cases: list[tuple[str, TestCase]] | None = None)[source]#

Bases: object

Bundles operations + fixture loader for declarative YAML tests.

Usage:

suite = DeclarativeTestSuite(
    operations={"normalize": normalize_fn},
    load_fixture=my_loader,
    expectations_dir="/path/to/expectations",
)

@suite.pytest_params()
def test_declarative(test_id, case):
    suite.run(test_id, case)

property cases: list[tuple[str, TestCase]]#: Return loaded test cases.

pytest_params()[source]#: Return pytest.mark.parametrize decorator with test IDs.

run(test_id: str, case: TestCase)[source]#: Execute a single declarative test case.

class gxformat2.testing.ExpectationSuite(*, root: dict[str, TestCase])[source]#

Bases: BaseModel

Top-level model: maps test IDs to test cases.

classmethod from_yaml(path: str) → ExpectationSuite[source]#: Load an expectation suite from a YAML file.

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

root: dict[str, TestCase]#

class gxformat2.testing.TestCase(*, fixture: str, operation: str, expect_error: bool = False, assertions: list[Assertion] = [])[source]#

Bases: BaseModel

A single declarative test case.

assertions: list[Assertion]#

expect_error: bool#

fixture: str#

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

operation: str#

gxformat2.testing.assert_value(obj: Any, expected: Any)[source]#: Assert exact equality.

gxformat2.testing.assert_value_any_contains(obj: Any, expected: str)[source]#: Assert that at least one element in obj contains expected as a substring.

gxformat2.testing.assert_value_contains(obj: Any, expected: str)[source]#: Assert that expected is a substring of obj.

gxformat2.testing.assert_value_falsy(obj: Any)[source]#: Assert that obj is falsy.

gxformat2.testing.assert_value_matches(obj: Any, pattern: str)[source]#: Assert that obj matches the given regex pattern.

gxformat2.testing.assert_value_set(obj: Any, expected_items: list)[source]#

Assert unordered set equality.

For frozensets of NamedTuples: converts each item via _asdict() and compares as sets of frozen key-value pairs. For frozensets of primitives: plain set comparison.

gxformat2.testing.assert_value_truthy(obj: Any)[source]#: Assert that obj is truthy.

gxformat2.testing.assert_value_type(obj: Any, expected_type: str)[source]#: Assert that obj is an instance of the named type.

gxformat2.testing.load_expectation_cases(expectations_dir: str) → Iterator[tuple[str, TestCase]][source]#: Yield (test_id, TestCase) for every expectation YAML in a directory.

gxformat2.testing.navigate(obj: Any, path: list[str | int | dict[str, Any]]) → Any[source]#: Walk an object by a list of path elements.

gxformat2.testing.run_assertion(obj: Any, assertion: Assertion)[source]#: Run a single assertion against a navigated object.

gxformat2.testing.run_declarative_case(case: TestCase, operations: dict[str, Callable[[...], Any]], load_fixture: Callable[[str], Any])[source]#

Execute one declarative test case.

Args:: case: TestCase with fixture, operation, and optionally assertions / expect_error operations: mapping of operation name to callable load_fixture: callable that takes a fixture name and returns the loaded workflow dict

gxformat2.validators module#

Shared pydantic-validator dispatch for workflow dict inputs.

Callers pass a parsed workflow dict and get a pydantic model back (or a ValidationError). Used by the declarative-operation runner and the schema-rule catalog runner.

gxformat2.validators.validate_format2(wf_dict)[source]#: Validate a Format2 workflow dict with the lax (open) schema.

gxformat2.validators.validate_format2_strict(wf_dict)[source]#: Validate a Format2 workflow dict with the strict (closed) schema.

gxformat2.validators.validate_native(wf_dict)[source]#: Validate a native Galaxy workflow dict with the lax (open) schema.

gxformat2.validators.validate_native_strict(wf_dict)[source]#: Validate a native Galaxy workflow dict with the strict (closed) schema.

gxformat2.validators.validator_for_fixture(fixture_name: str, strict: bool) → Callable[[dict], object][source]#

Pick the validator matching a fixture filename’s format.

.ga → native; anything else → format2. strict picks the closed-schema flavor that rejects unknown fields.

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