graph

etl_craft.core.graph

The same-pipeline task dependency graph: validation, waves, readiness and dead ends.

The graph holds a pipeline's tasks and the CFG_TASK_DEPENDENCY edges between them. Edges to tasks in other pipelines stay out of it, because they have no place in the pipeline's own order; each task records only how many it has, since its RUN_CONDITION counts them too. The cross-pipeline gate settles them at run time.

Given each task's status under the active run, the graph answers which tasks can start now (ready) and which never-run tasks can never start (unsatisfiable). A retry resumes: a task already SUCCESS or SKIPPED is never ready again.

RunState module-attribute

RunState = Mapping[int, TaskRunState]

Each task's state under the active run, by task id; a missing task has not run.

SelfDependencyError

Bases: GraphError

A task depends on itself.

CycleError

Bases: GraphError

The dependency edges form a cycle, so no order can run them.

__init__

__init__(cycle: tuple[int, ...]) -> None

Keep the cycle's task ids in the order they were followed.

UnknownTaskError

Bases: GraphError

An edge names a task id that is not among the graph's tasks.

TaskNode dataclass

A CFG_TASKS row, reduced to what the graph needs.

run_condition is None for ALL, the default; run_condition_count is read only under N. cross_pipeline_edge_count is how many of the task's dependencies are on tasks in other pipelines.

TaskEdge dataclass

A same-pipeline CFG_TASK_DEPENDENCY row: task_id waits on depends_on_task_id.

TaskRunState dataclass

What one dependency edge needs from its upstream's AUD_TASK_RUN_LOG row.

status is None while the task has no row under the active run. rows_written is what the attempt inserted, updated or deleted; None on a row recorded before it was kept, where target_count stands in for it.

wrote_rows property

wrote_rows: bool

Whether the attempt wrote rows, as a HAS_DATA dependency asks.

DependencyGraph

A validated, immutable same-pipeline task graph; build it with build_graph.

task_ids property

task_ids: tuple[int, ...]

The task ids, in the order they were supplied.

__init__

__init__(nodes: tuple[TaskNode, ...], edges: tuple[TaskEdge, ...]) -> None

Index the edges by the task that waits on them.

dependencies_of

dependencies_of(task_id: int) -> list[TaskEdge]

Return the edges task_id waits on.

downstream_of

downstream_of(task_id: int) -> set[int]

Return every task that waits on task_id, directly or through others.

waves

waves() -> list[list[int]]

Return the tasks in static waves: each wave depends only on earlier waves.

This is the guaranteed-safe order, not the earliest one: a task comes after every upstream, even when its ANY or N condition could let it start sooner. graph prints it, generated DAGs chain it, and operator repairs use its topological order. Task ids within a wave are sorted.

total_edge_count

total_edge_count(task_id: int) -> int

Return how many dependencies task_id has, cross-pipeline ones included.

required_edge_count

required_edge_count(task_id: int) -> int

Return how many of task_id's dependencies must be satisfied before it runs.

Every dependency under ALL, one under ANY (none when there are none), and run_condition_count under N.

satisfied_edge_count

satisfied_edge_count(task_id: int, run_state: RunState, cross_pipeline_satisfied: int | None = None) -> int

Return how many of task_id's dependencies are satisfied now.

cross_pipeline_satisfied is how many of its cross-pipeline dependencies the cross-pipeline gate has confirmed. None means they have not been evaluated, and they all count as satisfied: the scheduler has to consider the task for its gate to evaluate them, so counting them as unsatisfied would never start it.

ready

ready(run_state: RunState, cross_pipeline_satisfied: Mapping[int, int] | None = None) -> list[int]

Return the sorted ids of the tasks that can start now.

A task is ready when its own status allows another attempt (it has not run, or it FAILED) and at least required_edge_count of its dependencies are satisfied. An edge is satisfied when its upstream is:

• SUCCESS for a SUCCESS edge;
• FAILED for a FAILURE edge;
• terminal (SUCCESS, FAILED or SKIPPED) for an ALWAYS edge;
• SUCCESS with a positive target count for a HAS_DATA edge.

cross_pipeline_satisfied maps a task id to its confirmed cross-pipeline count; a task missing from it counts its cross-pipeline dependencies as satisfied (see satisfied_edge_count). That assumption alone never starts a task while one of its same-pipeline upstreams has not finished: otherwise an ANY task would start next to the upstream it waits on, find the cross-pipeline edge unsatisfied, and settle as SKIPPED just before the upstream would have satisfied it.

unsatisfiable

unsatisfiable(run_state: RunState, *, failures_final: bool = False) -> list[int]

Return the sorted ids of the never-run tasks that can never become ready in this run.

The caller records them SKIPPED. A dependency is lost for good only when its upstream is settled (SUCCESS or SKIPPED) and does not satisfy it; a FAILED upstream may still succeed on a retry, unless failures_final says no retry will come in this run, and an IN-PROGRESS one has not finished. Skipping a task settles it, which can doom its own dependents, so the search repeats until nothing changes. Cross-pipeline dependencies always count as still possible.

Only tasks with no status are reported; a task that already ran keeps its status.

build_graph

build_graph(tasks: Sequence[TaskNode], edges: Sequence[TaskEdge]) -> DependencyGraph

Validate CFG_ rows and build their DependencyGraph.

Raises UnknownTaskError or SelfDependencyError for a bad edge, GraphError for a duplicate task, an unknown dependency type or an invalid run condition, and CycleError when the edges loop. The Engine DB's CHECK constraints reject most of these on insert; the graph checks them again so it is safe to use without one. An N count larger than the task's number of dependencies spans two tables, so only the graph can reject it.