Core

Workflows

Workflow is the discriminated union NfPipeline | NfModule, selected on each entry's type: field. Internal plumbing (the graph, events, runners) is typed against Workflow; construct a concrete NfPipeline/NfModule directly, or build one from raw config data via validated_workflow.

NfPipeline

Bases: BaseWorkflow

A Nextflow pipeline step executed with 'nextflow run'.

NfModule

Bases: BaseWorkflow

A Nextflow module step executed with 'nextflow module run'.

WorkflowType

Bases: StrEnum

Discriminates the two workflow shapes a config's type: key selects between: a full nextflow run pipeline, or a single nextflow module run.

Globals and transitions

GlobalOptions

Bases: BaseModel

Settings applied across every workflow in a metapipeline (the config's globals: key). params is merged under each workflow's own params at run time, with the workflow's own values winning on key collision; profile/config_file likewise act as a fallback a workflow can override, not a value forced onto it.

nextflow_version_tuple property

nextflow_version_tuple: Optional[tuple[int, int, int, int]]

nextflow_version as (major, minor, patch, is_edge), or None if unset. Missing minor/patch components default to 0.

Transition

Bases: BaseModel

A directed source -> target edge between two workflow ids.

The legacy top-level transitions: representation; equivalent to the target workflow declaring from: source. Prefer from: in new configs.

The graph

MetaworkflowGraph

MetaworkflowGraph()

The in-memory representation of a metapipeline: a directed graph of Workflow nodes plus shared GlobalOptions.

Mutating methods (add_workflow, add_transition, ...) validate the change immediately and raise GraphValidationError/ConfigValidationError rather than leaving the graph in an invalid state -- a failed mutation has no effect. Wrap a batch of mutations in deferred_validation() to validate once at the end instead of after each call.

Construct one via nf_meta.core.config.from_file/validated_graph rather than composing an empty graph by hand, unless you're building it up programmatically from scratch.

Source code in src/nf_meta/core/graph.py

def __init__(self):
    self.G: nx.DiGraph = nx.DiGraph()
    self.global_options = GlobalOptions()
    self.config_version: str = CONFIG_VERSION_MAX
    self._events: list[Event] = []
    # None = not suspended. A set = suspended, accumulating the ids of
    # workflows touched while suspended, so deferred_validation's exit
    # can scope the batched dynamic check to only those (see validate()).
    self._validation_suspended: Optional[set[str]] = None

pop_events

pop_events()

Return and clear the events emitted by mutations since the last call.

Used by nf_meta.core.session.Session to drive undo/redo history; most consumers don't need to call this directly.

Source code in src/nf_meta/core/graph.py

def pop_events(self):
    """Return and clear the events emitted by mutations since the last call.

    Used by `nf_meta.core.session.Session` to drive undo/redo history;
    most consumers don't need to call this directly.
    """
    events = tuple(self._events)
    self._events = []
    return events

add_workflow

add_workflow(wf: Workflow)

Add a new workflow node.

Runs dynamic (network-backed) validation immediately unless inside deferred_validation(). Raises ConfigValidationError and leaves the graph unchanged if wf's id already has unresolved param references, or if dynamic validation reports an error.

Source code in src/nf_meta/core/graph.py

def add_workflow(self, wf: Workflow):
    """Add a new workflow node.

    Runs dynamic (network-backed) validation immediately unless inside
    `deferred_validation()`. Raises `ConfigValidationError` and leaves the
    graph unchanged if `wf`'s id already has unresolved param references,
    or if dynamic validation reports an error.
    """
    if self._validation_suspended is not None:
        self._validation_suspended.add(wf.id)
    else:
        diagnostics = run_dynamic_checks(wf, previous=None)
        if any(d.severity == Severity.ERROR for d in diagnostics):
            raise ConfigValidationError(diagnostics)
        log_warnings(diagnostics)

    self.G.add_node(wf.id, workflow=wf.model_copy())

    try:
        if wf.params and self._validation_suspended is None:
            self.validate_param_references(wf)
        self._emit(WorkflowAdded(wf))
    except (GraphValidationError, ConfigValidationError) as e:
        self.G.remove_node(wf.id)
        raise e

update_workflow

update_workflow(wf: Workflow)

Replace the workflow with wf.id with wf.

Same validation behavior as add_workflow. Raises ValueError if no workflow with that id exists yet -- use add_workflow to create one.

Source code in src/nf_meta/core/graph.py

def update_workflow(self, wf: Workflow):
    """Replace the workflow with `wf.id` with `wf`.

    Same validation behavior as `add_workflow`. Raises `ValueError` if no
    workflow with that id exists yet -- use `add_workflow` to create one.
    """
    # TODO: Skip update if node unchanged? -> adds junk to history otherwise
    if wf.id not in self.G.nodes:
        raise ValueError("Workflow has invalid id. Update unsuccessful!")

    old_wf = self.get_workflow_by_id(wf.id)

    if self._validation_suspended is not None:
        self._validation_suspended.add(wf.id)
    else:
        diagnostics = run_dynamic_checks(wf, previous=old_wf)
        if any(d.severity == Severity.ERROR for d in diagnostics):
            raise ConfigValidationError(diagnostics)
        log_warnings(diagnostics)
        if wf.params:
            self.validate_param_references(wf)

    self.G.nodes[wf.id]["workflow"] = wf.model_copy()
    self._emit(WorkflowUpdated(old_workflow=old_wf, new_workflow=wf))  # type: ignore

remove_workflow

remove_workflow(wf_id: str, recursive=False)

Remove the workflow wf_id and every transition touching it.

Raises ValueError if wf_id doesn't exist. Does not remove workflows that only reference the removed one via ${wf_id:params:...} -- those become dangling references, caught on their own next validation.

Source code in src/nf_meta/core/graph.py

def remove_workflow(self, wf_id: str, recursive=False):
    """Remove the workflow `wf_id` and every transition touching it.

    Raises `ValueError` if `wf_id` doesn't exist. Does not remove
    workflows that only reference the removed one via `${wf_id:params:...}`
    -- those become dangling references, caught on their own next
    validation.
    """
    if wf_id not in self.G.nodes:
        raise ValueError("Invalid wf_id")

    for edge in list(self.G.in_edges(wf_id)):
        self.remove_transition(*edge)

    for edge in list(self.G.out_edges(wf_id)):
        self.remove_transition(*edge)

    removed_wf = self.get_workflow_by_id(wf_id)
    self.G.remove_node(wf_id)
    self._emit(WorkflowRemoved(removed_wf))  # type: ignore

add_transition

add_transition(tr: Transition)

Add an edge marking tr.source as a predecessor of tr.target.

Raises GraphValidationError if either endpoint doesn't exist yet, or if adding the edge would introduce a cycle.

Source code in src/nf_meta/core/graph.py

def add_transition(self, tr: Transition):
    """Add an edge marking `tr.source` as a predecessor of `tr.target`.

    Raises `GraphValidationError` if either endpoint doesn't exist yet,
    or if adding the edge would introduce a cycle.
    """
    self.validate_edge(tr.source, tr.target)

    if self.G.has_edge(tr.source, tr.target):
        print(f"[warning] Edge already exists: {tr.source}->{tr.target}")

    self.G.add_edge(tr.source, tr.target, transition=tr.model_copy())
    self._emit(TransitionAdded(tr))

    # can only be validated after-the-fact with the node in the graph
    if self._validation_suspended is None:
        self.validate_graph_topology()

remove_transition

remove_transition(source: str, target: str)

Remove the edge source -> target, if present.

No-op (returns without raising) if the edge doesn't exist.

Source code in src/nf_meta/core/graph.py

def remove_transition(self, source: str, target: str):
    """Remove the edge `source -> target`, if present.

    No-op (returns without raising) if the edge doesn't exist.
    """
    tr = self.get_transition(source, target)
    if not tr:
        return

    self.G.remove_edge(source, target)
    self._emit(TransitionRemoved(tr))

    if self._validation_suspended is None:
        source_wf = self.get_workflow_by_id(source)
        self.validate_param_references(source_wf)  # type: ignore

update_global_options

update_global_options(glob: GlobalOptions)

Replace the graph's GlobalOptions wholesale. Not validated against individual workflows -- per-workflow effects (e.g. a shared profile) are only checked when that workflow is itself next validated.

Source code in src/nf_meta/core/graph.py

def update_global_options(self, glob: GlobalOptions):
    """Replace the graph's `GlobalOptions` wholesale. Not validated against
    individual workflows -- per-workflow effects (e.g. a shared `profile`)
    are only checked when that workflow is itself next validated."""
    old_globals = self.global_options
    self.global_options = glob
    self._emit(GlobalOptionsUpdated(new_globals=glob, old_globals=old_globals))

validate_param_references

validate_param_references(wf: Workflow)

Check that every ${wf_id:params:key} reference in wf.params points at a real, already-resolvable predecessor param.

Raises ConfigValidationError (carrying one Diagnostic per bad reference) if any reference targets an unknown workflow, a workflow that isn't actually a predecessor of wf, or a param with no value.

Source code in src/nf_meta/core/graph.py

def validate_param_references(self, wf: Workflow):
    """Check that every `${wf_id:params:key}` reference in `wf.params`
    points at a real, already-resolvable predecessor param.

    Raises `ConfigValidationError` (carrying one `Diagnostic` per bad
    reference) if any reference targets an unknown workflow, a workflow
    that isn't actually a predecessor of `wf`, or a param with no value.
    """

    def diagnostic(ref, message: str) -> Diagnostic:
        field = (
            f"params.{ref.source_key}"
            if isinstance(ref, WorkflowReference)
            else "params"
        )
        return Diagnostic(
            workflow_id=ref.source_wf_id,
            field=field,
            yaml_path=f"workflows.{ref.source_wf_id}.{field}",
            message=message,
            severity=Severity.ERROR,
        )

    diagnostics: list[Diagnostic] = []
    for ref in wf.field_refs:
        if ref.namespace != "params":
            continue  # output references not yet validated
        referenced_wf = self.get_workflow_by_id(ref.target_wf_id)
        if not referenced_wf:
            diagnostics.append(
                diagnostic(ref, f"Reference to unknown workflow: {ref.target_wf_id}")
            )
            continue

        if ref.target_wf_id not in list(self.G.predecessors(wf.id)):
            diagnostics.append(
                diagnostic(
                    ref,
                    f"Reference to workflow {ref.target_wf_id} that is not a predecessor of {wf.id}",
                )
            )
            continue

        if not referenced_wf.params or not ref.target_key:
            diagnostics.append(
                diagnostic(ref, f"No referencable params in workflow {referenced_wf.id}")
            )
            continue

        if not referenced_wf.params.get(ref.target_key):
            diagnostics.append(
                diagnostic(ref, f"Reference to unresolvable param: {ref.target_key}")
            )

    if diagnostics:
        raise ConfigValidationError(diagnostics)

validate_graph_topology

validate_graph_topology()

Raise GraphValidationError if the graph contains a cycle.

Source code in src/nf_meta/core/graph.py

def validate_graph_topology(self):
    """Raise `GraphValidationError` if the graph contains a cycle."""
    # Detect cycles
    if not nx.is_directed_acyclic_graph(self.G):
        cycle = nx.find_cycle(self.G)
        raise GraphValidationError(f"Workflow graph contains a cycle: {cycle}")

validate_edge

validate_edge(src, tgt)

Raise GraphValidationError if src or tgt isn't a node in the graph.

Source code in src/nf_meta/core/graph.py

def validate_edge(self, src, tgt):
    """Raise `GraphValidationError` if `src` or `tgt` isn't a node in the graph."""
    if src not in self.G or tgt not in self.G:
        raise GraphValidationError(f"Edge refers to nonexistent node: {src}->{tgt}")

validate

validate(touched: Optional[set[str]] = None)

Fully re-validate the graph: topology, edge endpoints, workflow ids, and every workflow's param references.

Pass touched (a set of workflow ids) to also run a single batched dynamic (network-backed) validation pass scoped to just those workflows -- this is what deferred_validation() uses on exit, rather than dynamic-validating the whole graph on every batch. Raises GraphValidationError/ConfigValidationError on the first failure found; does not collect all errors across the whole graph in one call.

Source code in src/nf_meta/core/graph.py

def validate(self, touched: Optional[set[str]] = None):
    """Fully re-validate the graph: topology, edge endpoints, workflow
    ids, and every workflow's param references.

    Pass `touched` (a set of workflow ids) to also run a single batched
    dynamic (network-backed) validation pass scoped to just those
    workflows -- this is what `deferred_validation()` uses on exit, rather
    than dynamic-validating the whole graph on every batch. Raises
    `GraphValidationError`/`ConfigValidationError` on the first failure
    found; does not collect all errors across the whole graph in one
    call.
    """
    self.validate_graph_topology()

    # Detect missing workflow nodes
    for src, tgt in self.G.edges():
        self.validate_edge(src, tgt)

    # Ensure workflow IDs are valid (non-empty)
    for n in self.G.nodes:
        if not isinstance(n, str) or not n:
            raise GraphValidationError("Workflow id must be a non-empty string.")

    for n in self.G.nodes:
        wf = self.get_workflow_by_id(n)
        self.validate_param_references(wf)  # type: ignore

    if touched:
        # Batched, concurrent dynamic check scoped to only the workflows
        # touched while suspended — not the whole graph. A node removed
        # later in the same suspended block is filtered out (nothing to
        # check).
        raw = [self.get_workflow_by_id(n) for n in touched if n in self.G.nodes]
        workflows = [wf for wf in raw if wf is not None]
        diagnostics = check_all(workflows)
        if any(d.severity == Severity.ERROR for d in diagnostics):
            raise ConfigValidationError(diagnostics)
        log_warnings(diagnostics)

get_workflow_by_id

get_workflow_by_id(id: str) -> Optional[Workflow]

Return the workflow with this id, or None if it doesn't exist.

Source code in src/nf_meta/core/graph.py

def get_workflow_by_id(self, id: str) -> Optional[Workflow]:
    """Return the workflow with this id, or `None` if it doesn't exist."""
    try:
        return self.G.nodes[id].get("workflow")
    except KeyError:
        return None

get_transition

get_transition(source: str, target: str) -> Optional[Transition]

Return the source -> target transition, or None if no such edge exists.

Source code in src/nf_meta/core/graph.py

def get_transition(self, source: str, target: str) -> Optional[Transition]:
    """Return the `source -> target` transition, or `None` if no such edge exists."""
    try:
        return self.G.edges[(source, target)].get("transition")
    except KeyError:
        return None

get_transitions

get_transitions() -> list[Transition]

Return every transition in the graph, in no particular order.

Source code in src/nf_meta/core/graph.py

def get_transitions(self) -> list[Transition]:
    """Return every transition in the graph, in no particular order."""
    return [self.get_transition(*e) for e in self.G.edges]  # type: ignore

get_workflows

get_workflows() -> list[Workflow]

Return every workflow in the graph, in no particular order.

Source code in src/nf_meta/core/graph.py

def get_workflows(self) -> list[Workflow]:
    """Return every workflow in the graph, in no particular order."""
    return [self.get_workflow_by_id(n) for n in self.G.nodes]  # type: ignore

get_workflows_sorted

get_workflows_sorted() -> list[Workflow]

Returns workflows in valid execution order.

Source code in src/nf_meta/core/graph.py

def get_workflows_sorted(self) -> list[Workflow]:
    """Returns workflows in valid execution order."""
    nodes_sorted = list(nx.topological_sort(self.G))
    workflows = [self.get_workflow_by_id(n) for n in nodes_sorted]
    return workflows  # type: ignore

subset_workflows

subset_workflows(start: Optional[str] = None, target: Optional[str] = None, workflows: Optional[list[Workflow]] = None)

Filter workflows (default: get_workflows_sorted()) down to the sub-DAG reachable from start and able to reach target (inclusive on both ends). Omit either bound to leave that side unrestricted.

Source code in src/nf_meta/core/graph.py

def subset_workflows(
    self,
    start: Optional[str] = None,
    target: Optional[str] = None,
    workflows: Optional[list[Workflow]] = None,
):
    """Filter `workflows` (default: `get_workflows_sorted()`) down to the
    sub-DAG reachable from `start` and able to reach `target` (inclusive
    on both ends). Omit either bound to leave that side unrestricted."""

    if workflows is None:
        workflows = self.get_workflows_sorted()

    if not len(workflows):
        return []

    if start:
        fw_reachable = nx.descendants(self.G, start) | {start}
        workflows = [wf for wf in workflows if wf.id in fw_reachable]

    if target:
        bw_reachable = nx.ancestors(self.G, target) | {target}
        workflows = [wf for wf in workflows if wf.id in bw_reachable]

    return workflows

get_start_workflow

get_start_workflow() -> Optional[Workflow]

Return the first order in topological sorting or None if no nodes exist

Source code in src/nf_meta/core/graph.py

def get_start_workflow(self) -> Optional[Workflow]:
    """
    Return the first order in topological sorting or None if no nodes exist
    """
    workflows_sorted = self.get_workflows_sorted()
    if not len(workflows_sorted):
        return None
    else:
        return workflows_sorted[0]

successors

successors(wf: Workflow) -> list[Workflow]

Return wf's direct successors (workflows with from: wf.id).

Source code in src/nf_meta/core/graph.py

def successors(self, wf: Workflow) -> list[Workflow]:
    """Return `wf`'s direct successors (workflows with `from: wf.id`)."""
    return [self.get_workflow_by_id(n) for n in self.G.successors(wf.id)]  # type: ignore

predecessors

predecessors(wf: Workflow) -> list[Workflow]

Return wf's direct predecessors (the workflows in wf.from).

Source code in src/nf_meta/core/graph.py

def predecessors(self, wf: Workflow) -> list[Workflow]:
    """Return `wf`'s direct predecessors (the workflows in `wf.from`)."""
    return [self.get_workflow_by_id(n) for n in self.G.predecessors(wf.id)]  # type: ignore

deferred_validation

deferred_validation()

Suspend per-operation validation; on exit, run a full graph validation plus a single batched, concurrent dynamic-check pass scoped to the workflows touched while suspended (not the whole graph).

Source code in src/nf_meta/core/graph.py

@contextmanager
def deferred_validation(self):
    """
    Suspend per-operation validation; on exit, run a full graph
    validation plus a single batched, concurrent dynamic-check pass
    scoped to the workflows touched while suspended (not the whole
    graph).
    """
    self._validation_suspended = set()
    try:
        yield self
    finally:
        touched = self._validation_suspended
        self._validation_suspended = None
        self.validate(touched)

Parsing, construction, and serialization

Free functions at the config-file-format boundary: parsing raw YAML-shaped dicts, constructing/validating individual workflows or a whole graph from them, and serializing a graph back to a dict or file.

from_file

from_file(cfg_file: Path, *, collect_errors: bool = False) -> MetaworkflowGraph

Read cfg_file as YAML and build+validate a MetaworkflowGraph from it. Reads the whole file into memory; makes no other side effects. See validated_graph for the validation contract and collect_errors.

Source code in src/nf_meta/core/config.py

def from_file(cfg_file: Path, *, collect_errors: bool = False) -> "MetaworkflowGraph":
    """Read `cfg_file` as YAML and build+validate a `MetaworkflowGraph` from
    it. Reads the whole file into memory; makes no other side effects. See
    `validated_graph` for the validation contract and `collect_errors`."""
    with open(cfg_file) as fh:
        data = yaml.safe_load(fh)
    return validated_graph(data, collect_errors=collect_errors)

to_file

to_file(g: MetaworkflowGraph, file: Path | str) -> None

Write g to file as YAML (via dump_config), prefixed with a # yaml-language-server: $schema=... comment pointing at the published JSON Schema for editor autocompletion. Overwrites file if it exists.

Source code in src/nf_meta/core/config.py

def to_file(g: "MetaworkflowGraph", file: Path | str) -> None:
    """Write `g` to `file` as YAML (via `dump_config`), prefixed with a
    `# yaml-language-server: $schema=...` comment pointing at the published
    JSON Schema for editor autocompletion. Overwrites `file` if it exists."""
    with open(file, "w") as fh:
        fh.write(f"# yaml-language-server: $schema={SCHEMA_URL}\n")
        yaml.safe_dump(dump_config(g), fh, sort_keys=False)

dump_config

dump_config(g: MetaworkflowGraph) -> dict

Project a MetaworkflowGraph back to the config's dict shape (the inverse of validated_graph): config_version, globals (omitted if empty), and workflows keyed by id with each workflow's own from: list reconstructed from the graph's edges. Always writes the current dict-keyed workflows: + per-workflow from: form, never the legacy list/top-level-transitions: shapes, even if the graph was loaded from one of those.

Source code in src/nf_meta/core/config.py

def dump_config(g: "MetaworkflowGraph") -> dict:
    """Project a `MetaworkflowGraph` back to the config's dict shape (the
    inverse of `validated_graph`): `config_version`, `globals` (omitted if
    empty), and `workflows` keyed by id with each workflow's own `from:` list
    reconstructed from the graph's edges. Always writes the current
    dict-keyed `workflows:` + per-workflow `from:` form, never the legacy
    list/top-level-`transitions:` shapes, even if the graph was loaded from
    one of those."""
    from_by_target: dict[str, list[str]] = {}
    for tr in g.get_transitions():
        from_by_target.setdefault(tr.target, []).append(tr.source)

    workflows_dict = {}
    for w in g.get_workflows():
        wf_dict = {k: v for k, v in w.model_dump_config().items()}
        if w.id in from_by_target:
            wf_dict["from"] = from_by_target[w.id]
        workflows_dict[w.id] = wf_dict

    config_dict = {
        "config_version": g.config_version,
        "globals": g.global_options.model_dump(exclude_none=True)
        if g.global_options
        else None,
        "workflows": workflows_dict,
    }

    if not config_dict["globals"]:
        del config_dict["globals"]

    return config_dict

validated_graph

validated_graph(data: dict, *, collect_errors: bool = False) -> MetaworkflowGraph

Construct a MetaworkflowGraph from a raw, parsed-YAML config dict.

Workflows can be declared as a dict keyed by id (workflows: {<id>: {...}}, the preferred form) or as a legacy list of definitions each carrying their own id. Transitions are declared per-workflow via from: <id> or from: [<id>, ...]; a top-level transitions: key is also still accepted, but mixing both styles in the same file raises rather than merging them.

Both the list-format workflows: and the top-level transitions: key are deprecated -- reading either logs a CLI warning. Fix by opening and saving the config in the editor, which always writes the current dict-keyed workflows: + per-workflow from: form.

By default (collect_errors=False) this fails fast, raising on the first problem found. Pass collect_errors=True for a best-effort pass that collects every problem instead of stopping at the first -- used by the CLI's validate command.

Source code in src/nf_meta/core/config.py

def validated_graph(data: dict, *, collect_errors: bool = False) -> "MetaworkflowGraph":
    """Construct a MetaworkflowGraph from a raw, parsed-YAML config dict.

    Workflows can be declared as a dict keyed by id (`workflows: {<id>: {...}}`,
    the preferred form) or as a legacy list of definitions each carrying their
    own `id`. Transitions are declared per-workflow via `from: <id>` or
    `from: [<id>, ...]`; a top-level `transitions:` key is also still accepted,
    but mixing both styles in the same file raises rather than merging them.

    Both the list-format `workflows:` and the top-level `transitions:` key are
    deprecated -- reading either logs a CLI warning. Fix by opening and saving
    the config in the editor, which always writes the current dict-keyed
    `workflows:` + per-workflow `from:` form.

    By default (`collect_errors=False`) this fails fast, raising on the first
    problem found. Pass `collect_errors=True` for a best-effort pass that
    collects every problem instead of stopping at the first -- used by the
    CLI's `validate` command.
    """
    if collect_errors:
        return _full_validated_graph(data)
    return _fast_validated_graph(data)

validated_workflow

validated_workflow(data: dict) -> Workflow

Construct and fully validate a single workflow from a flat, raw dict (id optional -- the model's own default_factory generates one if omitted). Runs the same tier-0/1 construction plus tier-2 dynamic checks a workflow gets as part of a full config build. Returns the validated Workflow, or raises GraphValidationError/ConfigValidationError.

Source code in src/nf_meta/core/config.py

def validated_workflow(data: dict) -> Workflow:
    """Construct and fully validate a single workflow from a flat, raw dict
    (`id` optional -- the model's own `default_factory` generates one if
    omitted). Runs the same tier-0/1 construction plus tier-2 dynamic checks
    a workflow gets as part of a full config build. Returns the validated
    Workflow, or raises `GraphValidationError`/`ConfigValidationError`.
    """
    if not data:
        raise GraphValidationError("No config data provided")
    if not isinstance(data, dict):
        raise GraphValidationError(
            f"Expected a flat mapping, got {type(data).__name__!r}"
        )
    wf, diagnostics = construct_workflow(data, data.get("id", "<unknown>"))
    if wf is None:
        raise ConfigValidationError(diagnostics)
    dynamic_diagnostics = run_dynamic_checks(wf, previous=None)
    errors = [d for d in dynamic_diagnostics if d.severity == Severity.ERROR]
    if errors:
        raise ConfigValidationError(errors)
    log_warnings(dynamic_diagnostics)
    return wf

validated_globals

validated_globals(data: dict) -> GlobalOptions

Construct and validate a GlobalOptions block from a flat, raw dict. Returns the validated GlobalOptions, or raises GraphValidationError/ConfigValidationError.

Source code in src/nf_meta/core/config.py

def validated_globals(data: dict) -> GlobalOptions:
    """Construct and validate a `GlobalOptions` block from a flat, raw dict.
    Returns the validated `GlobalOptions`, or raises
    `GraphValidationError`/`ConfigValidationError`.
    """
    if not data:
        raise GraphValidationError("No config data provided")
    if not isinstance(data, dict):
        raise GraphValidationError(
            f"Expected a flat mapping, got {type(data).__name__!r}"
        )
    try:
        return GlobalOptions(**data)
    except ValidationError as e:
        raise ConfigValidationError(
            diagnostics_from_pydantic_error(e, workflow_id=None)
        ) from e

Errors and diagnostics

Diagnostic dataclass

Diagnostic(workflow_id: str | None, field: str, message: str, severity: Severity, yaml_path: str | None = None)

One validation problem with a single field of a single workflow (or of globals, when workflow_id is None). The common vocabulary every validation tier (pydantic construction, dynamic/network checks, cross-workflow reference checks) reports through, so callers handle one shape regardless of where a problem originated.

Severity

Bases: StrEnum

A Diagnostic's severity: ERROR blocks (raises ConfigValidationError); WARNING is logged but doesn't stop construction.

ConfigValidationError dataclass

ConfigValidationError(errors: list[Diagnostic], graph_messages: list[str] = list())

Bases: Exception

A workflow/config-definition-level validation failure, whether from tier-0/1 (pydantic construction), tier-2 (network-backed) checks, or cross-workflow reference validation. Every consumer treats all of these identically, so they share one exception type.

GraphValidationError

Bases: Exception

A structural problem with the graph itself: a cycle, an edge to a nonexistent node, or a malformed workflows:/from:/transitions: shape. Distinct from ConfigValidationError, which carries per-field Diagnostics about individual workflow/globals content.