Idempotency & Safe Task Design

A task is idempotent when running it twice gives the same result as running it once. Retries, clears and backfills all run tasks again (chapter 12). So they are only safe on idempotent tasks.

Here is the problem in two lines. This load adds the revenue of 2026-01-10 to daily_revenue, and it ran twice:

INSERT INTO daily_revenue
SELECT :start, COALESCE(SUM(amount), 0) FROM orders
WHERE status = 'paid' AND order_date >= :start AND order_date < :end;

This chapter shows four safe ways to write, the clock mistake, atomic publishing, and a test that proves a task is safe.

Idempotency is decided by how a task writes. Pick the pattern from the shape of your data:

The most common is delete-insert. Both statements run in one transaction, so they succeed or fail together:

BEGIN;
DELETE FROM daily_revenue WHERE revenue_date = :start;
INSERT INTO daily_revenue
SELECT :start, COALESCE(SUM(amount), 0) FROM orders
WHERE status = 'paid' AND order_date >= :start AND order_date < :end;
COMMIT;

Without the transaction, a failure after the DELETE leaves the day empty until a retry succeeds.

When a record can change later, use an upsert on its key. Order 104 changed from paid to refunded between runs. After three runs, the copy still had one row, and it was refunded:

INSERT INTO orders_copy (order_id, status, amount)
SELECT order_id, status, amount FROM orders WHERE order_id = 104
ON CONFLICT (order_id) DO UPDATE
SET status = excluded.status, amount = excluded.amount;

A task must take its day from the run, not from the computer clock. Chapter 5 showed why: a retry the next day, or a rerun next week, reads a different day.

The clock hides in many places:

Search your DAGs and SQL for today(), now(), CURRENT_DATE and NOW(). Each hit is a bug that only shows up in a retry or a backfill.

Then test the task for an old date, and check that the output lands under that date, not today.

Idempotency asks: is running twice safe? Atomic publishing asks another question: what do readers see while the task runs?

TRUNCATE then INSERT empties the table first. During the load, the dashboard shows nothing. If the INSERT fails, the old data is gone too.

Instead, build the new table beside the old one, then swap the names in one transaction:

CREATE TABLE rev_new AS
SELECT order_date AS revenue_date, SUM(amount) AS revenue
FROM orders WHERE status = 'paid' GROUP BY order_date;

BEGIN;
ALTER TABLE rev RENAME TO rev_old;
ALTER TABLE rev_new RENAME TO rev;
COMMIT;

DROP TABLE rev_old;

The time between the build and the swap is the right place for checks. Count the rows, or compare totals with the source. If a check fails, do not swap.

A retry runs the whole task again. So a task that does five things repeats all five, even if only the last one failed.

Suppose one task loads orders, builds revenue and then emails the report. If the email fails, a retry loads and builds again. If the build fails after the email, a retry sends the email twice.

Split along those lines:

1. Split where steps fail for different reasons, such as a database and an API.
2. Split a slow step from a fast one, so a small failure does not repeat the slow work.
3. Put an action that cannot be undone, such as an email, in its own task, at the end, with retries=0.
4. Do not split steps that always succeed or fail together. Each task adds overhead.

When another system supports an idempotency key, build the key from the run, such as shop-report-2026-01-10. Then a repeated call does nothing the second time.

Reading the code is not enough. To prove a task is idempotent, run it twice for the same day and compare the table.

Here is the delete-insert load as an Airflow task:

@task
def load_revenue(data_interval_start=None, data_interval_end=None):
    day = data_interval_start.strftime("%Y-%m-%d")
    end = data_interval_end.strftime("%Y-%m-%d")
    conn = SqliteHook(sqlite_conn_id="shop_db").get_conn()
    with conn:                                   # one transaction
        conn.execute("DELETE FROM daily_revenue WHERE revenue_date = ?", (day,))
        conn.execute(INSERT_SQL, (day, day, end))
    print(f"loaded revenue for {day}")

A unit test can call the plain function twice. .function gives you the Python function behind a @task:

def test_load_revenue_twice_gives_the_same_table(...):
    start = pendulum.datetime(2026, 1, 10, tz="UTC")
    end = pendulum.datetime(2026, 1, 11, tz="UTC")
    query = ("SELECT count(*), sum(revenue) FROM daily_revenue "
             "WHERE revenue_date = '2026-01-10'")

    load_revenue.function(start, end)
    first = conn.execute(query).fetchone()
    load_revenue.function(start, end)
    second = conn.execute(query).fetchone()

    assert first == second == (1, 25.0)

You can do the same against a test Airflow. Run airflow tasks test twice for one old date, and run the same query after each run. In our run, both queries returned 1 row and 25.0.

Run it twice, get the same table:

1. Never append. Delete-insert one day, in one transaction.
2. Upsert on a key when records change later.
3. Take every date from the run, never from the clock.
4. Build aside and swap, so readers never see a half table.
5. Prove it: run twice, compare count and sum.

Check yourself: a load runs TRUNCATE, then INSERT, and the INSERT fails. What do readers see? (Answer: an empty table, and the old data is gone.)