Skip to main content

dag

Background DAG builder for v9 task templates.

This package converts a v9 FlowSpec (parsed from a task template YAML) into an executable background DAG.

Typical usage​

import desert
import yaml

from bitfount.flows.dag import DAGRunContext, build_background_dag, run_dag
from bitfount.flows.dag.context import FileMetadataContext
from bitfount.flows.dag.lifecycle import LifecycleNotifier
from bitfount.flows.schema import FlowSpec

# 1. Deserialise the v9 YAML into a FlowSpec.
with open("ga_trial_bronze_v9.yaml") as f:
    raw = yaml.safe_load(f)
flow_spec = desert.schema(FlowSpec).load(raw)

# 2. Parse + validate → BackgroundDAG.
dag = build_background_dag(flow_spec)

# 3. Build the run context.
run_ctx = DAGRunContext(
    context_providers={
        "file_metadata": FileMetadataContext(cache=cache, task_hash=task_hash)
    },
    runtime_params={"cache": cache, "task_hash": task_hash, "datasource": datasource},
    lifecycle_notifier=LifecycleNotifier(cache=cache, run_id=run_id),
)

# 4. Execute (fire-and-forget in a background thread / asyncio task).
await run_dag(dag, run_ctx)

Public API​

build_background_dag(flow_spec) Check the spec's declared version against this SDK's ceiling, then parse + validate a FlowSpec → BackgroundDAG. Defined in builder.py.

execute_dag(dag, run_ctx) Run all steps inside an already-active Prefect flow. Defined in executor.py.

run_dag(dag, run_ctx) Run all steps, creating a Prefect flow context if needed. Defined in executor.py.

build_flow(dag, run_ctx) Wrap execution in a named Prefect @flow for deployment scenarios. Defined in executor.py.

stamp_step_task_hashes(dag, datasource_hash) Assign each step its own cache partition key — a Merkle hash over its semantic config and its upstream steps' hashes — plus parent_task_hashes, the partition of each upstream step keyed by the consuming step's input parameter name. Called by the run-context builders in setup.py; defined in hashing.py.

Module​

Submodules​

Functions​

build_background_dag​

def build_background_dag(flow_spec: FlowSpec) ‑> BackgroundDAG:

Parse and validate a v9 FlowSpec into an executable BackgroundDAG.

This is the canonical entry point for building a background DAG. It combines parse_background_dag and validate_background_dag into a single call so callers do not need to orchestrate the two steps themselves.

Arguments

  • flow_spec: A deserialised FlowSpec (from bitfount.flows.schema). Must be a v9 flow — i.e. it must carry federation, roles, and worker fields.

Returns A BackgroundDAG ready to be passed to execute_dag.

Raises

  • FlowSpecVersionError: If flow_spec declares a task YAML version newer than this SDK knows, or one that cannot be parsed.
  • LegacyTaskYAMLError: If flow_spec is not a v9 flow.
  • StepNotFoundError: If any step's (task, version) is absent from the step registry.
  • ReferenceOrderError: If any FromRef input references a step declared later in the DAG (forward reference).
  • ValueError: If a step config cannot be validated against its Pydantic class.

build_flow​

def build_flow(    dag: RunnableDAG,    run_ctx: DAGRunContext,) ‑> collections.abc.Callable[[], collections.abc.Coroutine[typing.Any, typing.Any, dict[str, typing.Any]]]:

Return a named Prefect @flow callable for dag.

Useful for deployment scenarios where a first-class flow object is needed. For most runtime cases prefer run_dag.

Note that the flow's name is dag.name, which is shared by every project and datasource using the same template — it does not identify a run. Flow runs that need to be findable again are tagged at call time by _run_dag_flow_and_flush; see bitfount.runtimes.dedup.background_dag_tags.

Arguments

  • dag: A validated RunnableDAG (BackgroundDAG or InteractiveDAG).
  • run_ctx: Runtime context passed through to execute_dag.

Returns An async Prefect @flow callable with no required arguments.

build_interactive_dag​

def build_interactive_dag(    flow_spec: FlowSpec,) ‑> InteractiveDAG:

Parse and validate a v9 FlowSpec into an executable InteractiveDAG.

Canonical entry point for building the interactive (post-processing) DAG. Combines parse_interactive_dag and validate_interactive_dag.

Arguments

  • flow_spec: A deserialised v9 FlowSpec (must carry federation, roles, and worker).

Returns An InteractiveDAG ready to be executed (e.g. via build_flow, which runs any DAG-shaped object).

Raises

  • FlowSpecVersionError: If flow_spec declares a task YAML version newer than this SDK knows, or one that cannot be parsed.
  • LegacyTaskYAMLError: If flow_spec is not a v9 flow.
  • StepNotFoundError: If any step's (task, version) is absent from the step registry.
  • ReferenceOrderError: If any FromRef references a later interactive step (forward reference).
  • BackgroundStepNotFoundError: If any BackgroundRef references a step not declared in the background phase.
  • ValueError: If a step config cannot be validated against its Pydantic class.

execute_dag​

async def execute_dag(dag: RunnableDAG, run_ctx: DAGRunContext) ‑> dict[str, typing.Any]:

Run dag inside an already-active Prefect flow context.

Raises RuntimeError if no flow context is active — use run_dag when you cannot guarantee one.

Arguments

  • dag: A validated RunnableDAG (BackgroundDAG or InteractiveDAG).
  • run_ctx: Runtime context: providers, runtime params, reporter.

Returns Mapping of step name → step result for every executed step.

Raises

  • RuntimeError: If called outside a Prefect flow context.
  • Exception: Re-raises any step exception after calling run_ctx.lifecycle_notifier.on_failure.

run_dag​

async def run_dag(dag: RunnableDAG, run_ctx: DAGRunContext) ‑> dict[str, typing.Any]:

Run dag, creating a Prefect flow context if one is not already active.

Safe to call from anywhere — no existing flow context required.

Arguments

  • dag: A validated RunnableDAG (BackgroundDAG or InteractiveDAG).
  • run_ctx: Runtime context: providers, runtime params, reporter.

Returns Mapping of step name → step result for every executed step.

stamp_step_task_hashes​

def stamp_step_task_hashes(    dag: BackgroundDAG | InteractiveDAG,    datasource_hash: str,    resource_identities: Mapping[str, str] | None = None,    datastructure_identity: str | None = None,    datasource_type_identity: str | None = None,) ‑> dict[str, str]:

Compute and assign DAGStep.task_hash and .parent_task_hashes for dag.

Mutates the steps in place so each carries its own partition key: the executor injects it as the step's task_hash kwarg, and BackgroundResultsContext reads a background step's hash straight off the DAGStep it already holds — which is what keeps the two phases in agreement with no extra plumbing.

Each step also gets parent_task_hashes (see _resolve_parents), the same resolution the hashed payload used rather than a second one.

An interactive DAG's background_steps are hashed first and in the same pass, so a BackgroundRef's parent hash is the value the background phase computed. That holds only while both phases pass the same resource_identities and datastructure_identity; both derive them from the same pod config and FlowSpec, so they do.

Arguments

  • dag: A BackgroundDAG or InteractiveDAG. An InteractiveDAG's background_steps are hashed and stamped too.
  • resource_identities: Identity string per external resource kind, for steps declaring task_hash_resources.
  • datasource_hash: The datasource leaf hash — generate_prefect_task_hash(pod_name, datasource_name).
  • datastructure_identity: Canonical identity of the task's data_structure, for steps declaring a datastructure parameter.
  • datasource_type_identity: Identity of the datasource's type, for steps whose partition is keyed per file.

Returns Mapping of step name → assigned hash (interactive steps win on a name clash with a background step), for logging and assertions.

Classes​

BackgroundDAG​

class BackgroundDAG(**data: Any):

A fully parsed and validated background DAG ready for execution.

Attributes

  • name: Human-readable name of the flow (from FlowSpec.name).
  • federation_strategy: The federation strategy declared in the YAML (e.g. worker_only).
  • steps: Ordered list of steps. Execution order is declaration order; parallel steps are submitted as futures and flushed on demand.
  • extra: Any additional top-level metadata from the FlowSpec that callers may find useful (e.g. for logging or telemetry).

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static federation_strategy : str
  • static model_config
  • static name : str

BackgroundRef​

class BackgroundRef(**data: Any):

A cross-phase reference from an interactive step to a background step.

Background steps run in a separate execution; their outputs are read back from the cache at interactive-DAG runtime via BackgroundResultsContext.

Example YAML (inside an interactive step): ga_inference.cache

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static model_config
  • static output_field : str - Attribute name on the background step's result object (e.g. cache).
  • static step : str - Name of the background step whose result is referenced.

BackgroundStepNotFoundError​

class BackgroundStepNotFoundError(*args, **kwargs):

Raised when an interactive BackgroundRef targets an unknown step.

Interactive steps may reference background step outputs (e.g. ga_inference.cache). The referenced step must be declared in the flow's worker.background phase; otherwise its results can never be resolved from the cache.

ContextRef​

class ContextRef(**data: Any):

A reference to a value provided by an ambient ContextProvider.

Example YAML: $file_metadata.cache

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static field : str - Field name to resolve from the provider.
  • static model_config
  • static provider : str - Name of the context provider (e.g. file_metadata).

DAGRunContext​

class DAGRunContext(**data: Any):

Runtime context for a single background DAG execution.

Bundles everything the executor needs beyond the DAG itself so functions take (dag, run_ctx) instead of four separate parameters.

Attributes

  • context_providers: Mapping of provider name → ContextProvider used to resolve ContextRef inputs (e.g. file_metadata).
  • runtime_params: Runtime objects injected into every step's kwargs as a base layer (e.g. datasource, cache, task_hash). YAML-resolved refs override any overlapping keys.
  • lifecycle_notifier: Signals DAG lifecycle transitions (accept / success / failure) to whichever destinations are wired (cache Run row, initiator mailbox). Defaults to a no-op notifier.
  • background_results: Provider used to resolve BackgroundRef inputs (cross-phase references from interactive steps to background step outputs). None for background DAGs, which have no such refs.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static model_config

DAGStep​

class DAGStep(**data: Any):

A single step in the background DAG — one node to execute.

Attributes

  • name: Unique step name within the DAG (e.g. fovea_inference).
  • task: Registry key for the step (e.g. model_inference).
  • version: Step version to look up in the registry.
  • config: Instantiated Pydantic config for this step, or None if the step has no config class or no config was provided.
  • inputs: Mapping of parameter name → resolved InputRef.
  • parallel: If True the step is submitted as a Prefect future and its result is not awaited until a downstream step needs it.
  • save_to_cache: Declarative hint from the YAML. Parsed and stored but not currently acted upon — steps handle their own caching.
  • task_hash: This step's cache partition key — a Merkle hash over its own semantic config and its upstream steps' hashes, assigned by hashing.stamp_step_task_hashes when the run context is assembled. None until then (parsing and validation do not need it, and neither does a caller that only inspects the DAG); the executor then falls back to the run's datasource-level runtime_params["task_hash"].
  • parent_task_hashes: The task_hash of each upstream step this step declares an input from, keyed by this step's input parameter name — its routing table for reading an upstream partition through a direct store call, which a CacheAccessor cannot express. Assigned alongside task_hash; None until then. Keyed by parameter rather than upstream step name because the parameter is part of this step's own contract, while the upstream step's name is a label the flow author may change. Not part of the hashed payload (see hashing), so its contents never affect any partition key.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static config : pydantic.main.BaseModel | None
  • static model_config
  • static name : str
  • static parallel : bool
  • static parent_task_hashes : dict[str, str] | None
  • static save_to_cache : bool
  • static task : str
  • static task_hash : str | None
  • static version : int

DuplicateStepNameError​

class DuplicateStepNameError(*args, **kwargs):

Raised when two or more steps share the same name.

Node names are used as keys in execution result mappings and for FromRef(step=...) lookups. Duplicate names are therefore ambiguous and must be rejected at validation time.

FlowSpecVersionError​

class FlowSpecVersionError(*args, **kwargs):

Raised when a flow spec's declared version cannot be run by this SDK.

Covers three cases: the version is newer than this SDK's ceiling, the version cannot be read as a semantic version, or this SDK has no readable known versions to check against.

FromRef​

class FromRef(**data: Any):

A reference to a field on the result of a prior step.

Example YAML: ga_inference.cache

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static model_config
  • static output_field : str - Attribute name on the step's result object.
  • static step : str - Name of the step whose result is referenced.

InteractiveDAG​

class InteractiveDAG(**data: Any):

A fully parsed and validated interactive DAG ready for execution.

The interactive DAG is the post-processing phase of a v9 flow (GA calculation, criteria matching, report generation). It runs as a separate execution from the background DAG and may reference background step outputs via BackgroundRef inputs, which are resolved from the cache at runtime.

Attributes

  • name: Human-readable name of the flow (from FlowSpec.name).
  • federation_strategy: The federation strategy declared in the YAML (e.g. worker_only).
  • steps: Ordered list of interactive steps. Execution order is declaration order; parallel steps are submitted as futures and flushed on demand.
  • background_steps: The parsed background DAGStep list (carrying each step's instantiated config). Used to build the BackgroundResultsContext and to validate BackgroundRef targets; not executed by the interactive run.
  • extra: Any additional top-level metadata from the FlowSpec.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Variables​

  • static background_steps : list[DAGStep]
  • static federation_strategy : str
  • static model_config
  • static name : str
  • background_step_names : frozenset[str] - Names of the background steps interactive steps may reference.

LegacyTaskYAMLError​

class LegacyTaskYAMLError(*args, **kwargs):

Raised when a non-v9 FlowSpec is passed to the background DAG parser.

v9 flows are identified by the presence of federation, roles, and worker fields. Older task YAML formats (v8 and below) use a flat ModellerConfig structure and are not supported by this parser.

ReferenceOrderError​

class ReferenceOrderError(*args, **kwargs):

Raised when a FromRef references a step declared later in the DAG.

The background DAG is executed in declaration order. A step cannot consume the output of a step that has not yet run.

StepNotFoundError​

class StepNotFoundError(*args, **kwargs):

Raised when a DAG step references a task that is not in the registry.

Includes the task name and version in the message so the user can identify which YAML entry needs correcting.