Homework 06 — Idempotency in practice

Task 1 — Make the daily rollup idempotent

In Lesson 05's homework you added dags/gharchive_daily_rollup.py with a _rollup function that wrote the output directly via out.to_parquet(final_path). This is not atomic — a mid-write crash leaves a partial file at the final path.

Refactor _rollup to use the staging-then-rename pattern from _helpers.py:

• Compute the final path as before.
• Use transactional_write_path(final, context["run_id"]) to get a staging path.
• Write output to the staging path with out.to_parquet(staging, index=False).
• Call atomic_replace(staging, final) to finish the write.

Self-check: trigger a manual run and kill the scheduler mid-write (same technique as Lab 06 Step 5). Inspect the daily_rollups/ directory — you should see a *.staging.* file but no final .parquet. After clearing and re-running, the staging file is gone and the final file is correct. Re-running the same day a second time produces the same file with no duplicates.

Task 2 — Tune retry budgets with a failure-notification task

Add a task notify_slack that runs after transform but only when transform fails. Use BashOperator with the command echo "would notify slack: the hour {{ ds }} transform failed". Set trigger_rule="one_failed" and retries=0 on the operator.

Then deliberately fail transform by temporarily adding raise RuntimeError("injected failure") at the top of _transform. Trigger a manual run.

Self-check: in the Grid view, transform is red (failed) and notify_slack is green (success). The task log for notify_slack shows exactly one echo line. Remove the injected error before moving on.

Task 3 — Identify the non-idempotent operation

Below are four hypothetical task bodies. For each one, state whether it is idempotent. For the non-idempotent ones, name the specific fix.

(a)

def task(**ctx):
    with open("counter.txt", "a") as f:
        f.write("1\n")

(b)

def task(**ctx):
    db.execute(
        "INSERT INTO events (hour, count) VALUES (%s, %s)",
        (hour, n),
    )

(c)

from uuid import uuid4

def task(**ctx):
    requests.post(
        "https://api.example.com/orders",
        json={"id": uuid4().hex, "amount": 100},
    )

(d)

def task(**ctx):
    df.to_parquet(f"out/{hour}.parquet")  # pandas default: overwrite

Self-check: you can articulate — in one sentence each — which operations are idempotent, which are not, and the minimal change that would make the non-idempotent ones safe.