Skip to main content

executor

Executor: runs a RunnableDAG inside a Prefect flow.

This is the runtime heart of the DAG system. It takes a fully parsed and validated RunnableDAG (a BackgroundDAG or InteractiveDAG) and executes every step in declaration order, wiring inputs from prior step results or context providers.

Parallel execution​

Nodes marked parallel=True are run concurrently via asyncio.to_thread, which schedules the (synchronous) task function on the default thread-pool executor. Results stay in-process — no Prefect serialisation boundary is crossed — so private attributes such as _accessor on cache-backed result types are preserved for downstream steps. Tasks are tracked as asyncio.Task objects in a pending dict; they are awaited lazily when a downstream step declares a FromRef dependency, or flushed at the end of the loop.

Runtime parameters​

Steps often require runtime objects (datasource, cache, task_hash, …) that are not declared in the YAML. Pass them via DAGRunContext.runtime_params; they are merged into every step's kwargs before YAML-resolved inputs, so YAML refs can override them when needed.

Public API​

execute_dag(dag, run_ctx) Runs the DAG. Caller must already be inside a Prefect @flow context; raises RuntimeError otherwise.

run_dag(dag, run_ctx) Ensures a Prefect flow context exists, then delegates to execute_dag. Safe to call from anywhere.

build_flow(dag, run_ctx) Returns a named Prefect @flow callable for deployment scenarios.

build_background_flow(dag, task_hash, project_id, open_run) Returns a @flow callable that opens the run inside the flow body, so a background run's flow run is the outermost boundary of its work.

Module​

Functions​

build_background_flow​

def build_background_flow(    dag: BackgroundDAG,    *,    task_hash: str,    project_id: str,    open_run: Callable[[], DAGRunContext | None],) ‑> collections.abc.Callable[[], collections.abc.Coroutine[typing.Any, typing.Any, dict[str, typing.Any] | prefect.client.schemas.objects.State]]:

Return a @flow callable that opens its own run, inside the flow.

The background counterpart to build_flow, and the shape that makes "a Prefect flow run exists" a total invariant for background work. build_flow needs a DAGRunContext in hand, which means the run row, the datasource materialisation and the model resolution all happen before anything is observable; a crash in that window leaves a running cache row and no flow run to explain it. Here open_run is deferred into the flow body, so all of it is inside the boundary: a setup failure is a failed flow run rather than a log line and a dropped trigger.

The duplicate guard lives here too, for the same reason — it is now Prefect's flow-run state, not the cache's running row, that decides whether a second run may start (see bitfount.runtimes.dedup.is_background_flow_run_active, which is stale-aware so a dead run cannot wedge the lineage). A duplicate ends as an immediately-terminal SKIPPED_STATE_NAME run naming the live run it duplicates, which is visible in the Prefect UI where the old early return None was visible nowhere.

Arguments

  • dag: The validated background DAG to execute.
  • task_hash: The (pod, datasource) task hash, for the duplicate guard.
  • project_id: Project the run belongs to, for the duplicate guard.
  • open_run: Deferred run-context assembly — typically functools.partial(build_background_run_context, ...). Called once, inside the flow body, after the guard has passed. None means the caller declined the run for its own reasons, and is reported the same way a duplicate is.

Returns An async Prefect @flow callable with no required arguments. It returns the step results, or a terminal State when nothing ran.

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.

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.

run_flow_resiliently​

async def run_flow_resiliently(flow_fn: Callable[[], Awaitable[Any]]) ‑> Any:

Run a Prefect flow, retrying the transient Windows start-time 409.

Guards against the Windows-only flow_run_state timestamp collision (see the module note above): a bounded retry re-creates the flow run a clock-tick later so its Pending/Running state timestamps differ. Only a 409 is retried; any other error — and the final 409 after exhausting retries — is re-raised unchanged. Safe against re-running work: the collision happens at begin_run before any step executes (a real DAG does far more than one clock tick of work, so the Running->Completed transition never shares a tick).

Each retry creates a new flow run, so the run it gave up on is left behind at PENDING, carrying this run's lineage tags. Two consequences, handled separately because they need different remedies:

  • The duplicate guard must not treat a pre-RUNNING run as live, or the retry would find the run it just abandoned and skip itself as a duplicate — see bitfount.runtimes.dedup.BACKGROUND_ACTIVE_STATE_TYPES. That is correctness, and it does not depend on the cleanup below succeeding.
  • The abandoned run is ended as CANCELLED here (_abandon_flow_run), so it does not sit PENDING forever misrepresenting the pod's history. That is tidiness, and it is best-effort.

Global variables​

  • ABANDONED_STATE_NAME - State name given to the flow run a 409 retry left behind. The type is CANCELLED because that is what happened: the run was created and then never executed, by our decision rather than by a failure of its own.
  • SKIPPED_STATE_NAME - State name given to a background flow run that stopped because another run for the same lineage was already executing. The state type is COMPLETED, so the run is terminal, drops straight out of the liveness set, and raises nothing at the caller — a duplicate trigger is normal operation, not a failure. The name is what keeps it honest in the UI: nothing ran.