Lab 1 — Build a tiny DAG runner

Objective

By the end of this lab you will have a runnable Python module, dagrunner.py, that is a working model of Airflow's core scheduler. It lets you define tasks with dependencies, resolves a valid execution order with a topological sort, runs the tasks in that order, retries a failing task up to a cap, skips tasks whose upstreams failed, and writes a run.log recording every attempt. It uses only the Python standard library — no Airflow, no Docker, no network — so it costs nothing and runs anywhere.

This is not a toy for its own sake. The loop you build here — "run a task only once its dependencies have succeeded, and retry on failure" — is the exact responsibility the Airflow scheduler carries in production. Building it once is what makes the real scheduler legible.

Skills exercised:

• Modeling a DAG as tasks plus dependency edges in Python
• Implementing a topological sort with cycle detection
• Executing tasks in dependency order with bounded retry-on-failure
• Recording run outcomes to a durable log for later inspection

Prerequisites and setup

Prerequisites: the three lessons in this module — Why orchestration exists, Airflow's core model, and Operating pipelines in production. This lab makes the scheduler loop from those lessons concrete.

Setup: you need Python 3.10 or newer. There are no packages to install; everything uses the standard library. Verify your version and create a working folder.

macOS / Linux

python3 --version
mkdir de204-dag-runner-lab && cd de204-dag-runner-lab

Windows (PowerShell)

python --version
mkdir de204-dag-runner-lab; cd de204-dag-runner-lab

Expected output from the version command is a line like Python 3.11.6. If the command is not found, install Python from python.org before continuing. In the steps below, use python3 on macOS/Linux and python on Windows.

note

Every step edits the same dagrunner.py in this one folder. If you close your terminal and come back, re-run cd de204-dag-runner-lab to return to it; your file is still there. Each step adds to dagrunner.py, so keep everything you have already written.

Step 1 — Model tasks and a DAG

Goal: create the data structures that hold a graph — a Task with a retry cap, and a DAG that stores tasks and the upstream edges between them.

Create dagrunner.py with these two classes:

from collections import defaultdict, deque


class Task:
    """A unit of work: an id, a zero-argument callable, and a retry cap."""

    def __init__(self, task_id, run, retries=0):
        self.task_id = task_id
        self.run = run
        self.retries = retries  # extra attempts allowed after the first


class DAG:
    """A named set of tasks plus the upstream edges between them."""

    def __init__(self, dag_id):
        self.dag_id = dag_id
        self.tasks = {}
        self.upstream = defaultdict(set)  # task_id -> set of upstream task_ids

    def add_task(self, task):
        self.tasks[task.task_id] = task
        self.upstream.setdefault(task.task_id, set())

    def set_dependency(self, upstream_id, downstream_id):
        """Declare that downstream_id runs only after upstream_id."""
        self.upstream[downstream_id].add(upstream_id)

Checkpoint: build a two-task DAG and confirm the edge is recorded.

python3 -c "import dagrunner as d; \
g = d.DAG('demo'); \
g.add_task(d.Task('extract', lambda: None)); \
g.add_task(d.Task('load', lambda: None)); \
g.set_dependency('extract', 'load'); \
print(sorted(g.tasks), dict(g.upstream))"

You should see ['extract', 'load'] {'extract': set(), 'load': {'extract'}}. The load task lists extract as upstream; extract has an empty upstream set. If upstream is empty, check that set_dependency was called with ('extract', 'load') in that order.

Step 2 — Resolve execution order with a topological sort

Goal: turn the graph into a runnable order where every task appears after all of its upstream tasks — and raise if the graph has a cycle, because a cycle has no valid order.

Add this function to dagrunner.py. It uses Kahn's algorithm: repeatedly take a task with no unmet dependencies, and process its downstream edges.

def topological_order(dag):
    """Return task ids in a runnable order, or raise if the graph has a cycle."""
    indegree = {task_id: 0 for task_id in dag.tasks}
    downstream = defaultdict(set)
    for down, ups in dag.upstream.items():
        for up in ups:
            downstream[up].add(down)
            indegree[down] += 1

    ready = deque(sorted(t for t, deg in indegree.items() if deg == 0))
    order = []
    while ready:
        current = ready.popleft()
        order.append(current)
        for down in sorted(downstream[current]):
            indegree[down] -= 1
            if indegree[down] == 0:
                ready.append(down)

    if len(order) != len(dag.tasks):
        raise ValueError("DAG has a cycle; cannot schedule")
    return order

Sorting the ready tasks keeps the order deterministic, so two runs of the same graph produce the same sequence. The cost is linear in tasks plus edges, $O(V + E)$.

Checkpoint: confirm a valid order on a chain, then that a cycle raises.

python3 -c "import dagrunner as d; \
g = d.DAG('chain'); \
[g.add_task(d.Task(t, lambda: None)) for t in ('a', 'b', 'c')]; \
g.set_dependency('a', 'b'); g.set_dependency('b', 'c'); \
print('order:', d.topological_order(g))"

The first command prints order: ['a', 'b', 'c']. Now feed it a cycle and confirm it refuses to schedule:

python3 -c "import dagrunner as d; \
c = d.DAG('cycle'); \
[c.add_task(d.Task(t, lambda: None)) for t in ('x', 'y')]; \
c.set_dependency('x', 'y'); c.set_dependency('y', 'x'); \
d.topological_order(c)"

This second command intentionally fails: the last line of the traceback reads ValueError: DAG has a cycle; cannot schedule. Seeing that error is the intended result, not a problem. If the chain order is not a, b, c, recheck the two set_dependency calls; if the cycle does not raise, confirm the final len(order) != len(dag.tasks) guard is present.

Step 3 — Run a task with retry-on-failure

Goal: execute one task, re-attempting it on failure up to its cap, and report how many attempts it took and whether it ended in success.

Add this function. A task with retries=2 is allowed three attempts total: the first, plus two retries.

def run_task(task):
    """Run a task, retrying on failure up to its cap. Return (attempts, state)."""
    attempt = 0
    while True:
        attempt += 1
        try:
            task.run()
            return attempt, "success"
        except Exception as exc:
            if attempt <= task.retries:
                continue
            return attempt, f"failed: {exc}"

Checkpoint: run a task that fails once then succeeds, and one that always fails, and inspect the attempt counts.

python3 -c "import dagrunner as d; \
counter = []; \
flaky = lambda: counter.append(1) or ( \
    (_ for _ in ()).throw(RuntimeError('reset')) if len(counter) < 2 else None); \
print('flaky:', d.run_task(d.Task('f', flaky, retries=2))); \
boom = lambda: (_ for _ in ()).throw(RuntimeError('disk full')); \
print('always:', d.run_task(d.Task('b', boom, retries=1)))"

You should see flaky: (2, 'success') — it failed once, then succeeded on the second attempt — and always: (2, 'failed: disk full') — two attempts (first plus one retry), then it gives up. If flaky shows (1, 'success'), the raise on the first attempt is not firing; if always shows three attempts, check that the cap is attempt <= task.retries.

Step 4 — Run the whole DAG and write a run log

Goal: tie it together. Walk the topological order, run each task, skip any task whose upstream failed, and write a durable run.log — the metadata record a real scheduler keeps.

Add this function:

def run_dag(dag, log_path):
    """Execute the DAG in order, skipping tasks whose upstreams failed."""
    order = topological_order(dag)
    failed = set()
    log_lines = []
    for task_id in order:
        blockers = dag.upstream[task_id] & failed
        if blockers:
            failed.add(task_id)
            log_lines.append(
                f"{task_id}\tupstream_failed\tattempts=0\t"
                f"blocked_by={','.join(sorted(blockers))}"
            )
            continue
        attempts, state = run_task(dag.tasks[task_id])
        if state.startswith("failed"):
            failed.add(task_id)
            log_lines.append(f"{task_id}\tfailed\tattempts={attempts}\t{state}")
        else:
            log_lines.append(f"{task_id}\tsuccess\tattempts={attempts}")

    with open(log_path, "w", newline="") as f:
        f.write("\n".join(log_lines) + "\n")
    return {"order": order, "failed": failed, "log_lines": log_lines}

Checkpoint: run a graph where the middle task always fails, and confirm the downstream task is skipped rather than run.

python3 -c "import dagrunner as d; \
g = d.DAG('skip'); \
g.add_task(d.Task('a', lambda: None)); \
boom = lambda: (_ for _ in ()).throw(RuntimeError('disk full')); \
g.add_task(d.Task('b', boom, retries=1)); \
g.add_task(d.Task('c', lambda: None)); \
g.set_dependency('a', 'b'); g.set_dependency('b', 'c'); \
r = d.run_dag(g, 'skip.log'); \
print('failed:', sorted(r['failed'])); \
print(open('skip.log').read().strip())"

You should see failed: ['b', 'c'], then three log lines: a succeeded, b failed after 2 attempts, and c is upstream_failed with blocked_by=b. The skip is the point — c never runs on a failed upstream, exactly as an orchestrator holds back downstream tasks. If c shows success, the blockers = dag.upstream[task_id] & failed guard is not catching it.

Step 5 — Assemble a sample DAG with a flaky task

Goal: give the runner a realistic graph to prove itself on — a daily-sales pipeline whose load task fails on its first attempt and succeeds on a retry, mirroring the transient failures from the production lesson.

Add this builder and a runnable entry point to finish dagrunner.py:

def build_sales_dag():
    """A four-task daily-sales DAG with one deliberately flaky task."""
    events = []

    def extract():
        events.append("extract")

    def transform():
        events.append("transform")

    attempts_seen = {"load": 0}

    def load():
        attempts_seen["load"] += 1
        events.append(f"load#{attempts_seen['load']}")
        if attempts_seen["load"] < 2:
            raise RuntimeError("warehouse connection reset")

    def notify():
        events.append("notify")

    dag = DAG("daily_sales")
    dag.add_task(Task("extract", extract))
    dag.add_task(Task("transform", transform))
    dag.add_task(Task("load", load, retries=2))
    dag.add_task(Task("notify", notify))
    dag.set_dependency("extract", "transform")
    dag.set_dependency("transform", "load")
    dag.set_dependency("load", "notify")
    return dag, events


if __name__ == "__main__":
    dag, events = build_sales_dag()
    result = run_dag(dag, "run.log")
    print("order:", result["order"])
    print("failed:", sorted(result["failed"]))
    print("events:", events)

Checkpoint: run the whole file, then read the log it wrote.

python3 dagrunner.py
cat run.log

The program prints an order of ['extract', 'transform', 'load', 'notify'], an empty failed: [], and events showing load#1 then load#2 (the retry). The run.log then contains four lines, with load recorded as success at attempts=2. On Windows PowerShell, use Get-Content run.log instead of cat.

Validation

Verify the whole build against the objective: a runner that orders by dependency, retries the flaky load, and logs every attempt. Run this validation one-liner, which reruns the sample DAG and checks the order, the retry count, and that nothing failed.

python3 -c "import dagrunner as d; \
dag, events = d.build_sales_dag(); \
r = d.run_dag(dag, 'run.log'); \
rows = [line.split(chr(9)) for line in open('run.log').read().splitlines()]; \
state = {row[0]: (row[1], row[2]) for row in rows}; \
ok = (r['order'] == ['extract', 'transform', 'load', 'notify'] \
      and not r['failed'] \
      and state['load'] == ('success', 'attempts=2') \
      and all(v[0] == 'success' for v in state.values())); \
print('PASSED - dag runner complete' if ok else 'FAILED - inspect run.log')"

Checkpoint: the command prints PASSED - dag runner complete. If it prints FAILED, open run.log: load must read success with attempts=2, and all four tasks must be success. A wrong order points back to Step 2; a load that did not reach attempts=2 points back to the retry cap in Step 3.

Teardown

Nothing is billable — this lab runs entirely on your machine. Keep the de204-dag-runner-lab folder if you want to reuse dagrunner.py as a reference for the rest of the course; otherwise delete it with rm -rf de204-dag-runner-lab (macOS/Linux) or Remove-Item -Recurse de204-dag-runner-lab (Windows).

Troubleshooting

Symptom	Likely cause	Fix
ModuleNotFoundError: No module named 'dagrunner'	Running from the wrong folder	cd into de204-dag-runner-lab, where dagrunner.py lives
ValueError: DAG has a cycle; cannot schedule	A dependency loop in your graph	Remove the back-edge; upstream must never depend on its downstream
Topological order is missing a task	Task added only via set_dependency	Every task must be registered with add_task before running
load shows attempts=1 in Step 5	Flaky task not raising on attempt 1	Confirm attempts_seen['load'] < 2 raises before the second run
Downstream task runs after an upstream failed	Skip guard missing in run_dag	Keep the dag.upstream[task_id] & failed check before running
Validation prints FAILED	Order or retry count wrong	Order comes from Step 2; the attempts=2 retry comes from Step 3