🧭 Concept Pipelines & automation Automation & Pipelines · step 3 of 26 Updated

Idempotency Explained: Why a GIS Job Must Be Safe to Re-run

Problem statement

The nightly job failed at 02:47. You re-run it at 09:15 and make things worse:

$ python monthly_load.py
loaded 48,201 parcels

$ python monthly_load.py        # after fixing the source file
loaded 48,201 parcels

$ psql -c "select count(*) from parcels"
 96402

Or the file version:

data/out/
  parcels_2026-08-11.gpkg
  parcels_2026-08-11_1.gpkg
  parcels_2026-08-11_2.gpkg      # which one is the real one?

The job is not idempotent: running it twice does not leave the same state as running it once. That single property decides whether a failure is a two-minute retry or a manual clean-up, whether a scheduler may retry automatically, and whether anyone can safely re-run last month to reproduce a number.

Quick answer

Make every effect replaceable rather than additive:

  1. Replace, do not append — overwrite the output, or delete-then-insert within a key
  2. Derive output names deterministically — same inputs, same paths, no counters or timestamps-of-now
  3. Write atomically — to a temporary name, then rename, so a failure leaves no partial output
  4. Make "already done" cheap to detect — a marker, a checksum, or a state file
  5. Take time as an input — a run date parameter, never datetime.now() inside the logic
from pathlib import Path
import geopandas as gpd

def load(gdf: gpd.GeoDataFrame, run_date: str) -> Path:
    dest = Path(f"data/out/parcels_{run_date}.gpkg")     # deterministic from an input
    dest.parent.mkdir(parents=True, exist_ok=True)

    tmp = dest.with_suffix(".gpkg.part")
    gdf.to_file(tmp, driver="GPKG")                       # write aside
    tmp.replace(dest)                                     # atomic replace
    return dest

# running this twice with the same run_date produces exactly one file,
# with exactly the contents of the second run

The mental test: if the job is killed at any point and restarted, is the world still consistent? If yes, it is idempotent; if you have to check what state things are in first, it is not.

What "safe to re-run" means

Panels contrasting an additive job that doubles data on re-run with a replacing job that converges.
Same code path, twice — one converges on the right state, the other accumulates.

Step-by-step solution

Grid of side effects and the idempotent pattern for each: files, databases, APIs, notifications.
Every kind of side effect has an idempotent form — and a naive form that breaks on retry.

Files: replace, and make the name a function of the inputs 

from pathlib import Path

# not idempotent — a new file every run
dest = Path(f"data/out/parcels_{datetime.now():%Y%m%d_%H%M%S}.gpkg")

# not idempotent — versions accumulate
n = 1
while dest.exists():
    dest = dest.with_stem(f"{dest.stem}_{n}"); n += 1

# idempotent — the same inputs always produce the same path, and it is replaced
dest = Path(f"data/out/parcels_{run_date}.gpkg")
gdf.to_file(dest, driver="GPKG")

The rule is that the output path must be derivable from the inputs and parameters, not from the clock at the moment of writing. Then a second run targets the same path and replaces it.

Atomic writes: never leave a half-file 

from pathlib import Path
import geopandas as gpd

def write_atomic(gdf, dest: Path, driver="GPKG") -> Path:
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(dest.name + ".part")
    gdf.to_file(tmp, driver=driver)
    tmp.replace(dest)          # atomic rename on the same filesystem
    return dest

Without this, a crash mid-write leaves a file that exists but is truncated — and every downstream check that treats existence as "done" is now wrong. With it, the destination only ever appears complete.

Databases: delete-then-insert inside a transaction 

from sqlalchemy import text
import geopandas as gpd

def load_partition(gdf, engine, table: str, run_date: str) -> int:
    """Replace exactly this run's slice — safe to repeat."""
    with engine.begin() as conn:                      # one transaction
        conn.execute(text(f"DELETE FROM {table} WHERE run_date = :d"), {"d": run_date})
        gdf.assign(run_date=run_date).to_postgis(table, conn, if_exists="append", index=False)
        count = conn.execute(
            text(f"SELECT count(*) FROM {table} WHERE run_date = :d"), {"d": run_date}
        ).scalar()
    return count

The pattern is delete the slice this run owns, then insert it, both inside one transaction. Re-running replaces the same slice; a failure rolls back and leaves the previous state. if_exists="replace" also works when the run owns the whole table.

An upsert achieves the same at row level:

INSERT INTO parcels (parcel_id, class, geom, updated_at)
VALUES (:parcel_id, :class, ST_GeomFromWKB(:geom, 27700), :updated_at)
ON CONFLICT (parcel_id) DO UPDATE
SET class = EXCLUDED.class, geom = EXCLUDED.geom, updated_at = EXCLUDED.updated_at;

Make "already done" detectable 

import hashlib, json
from pathlib import Path

STATE = Path("logs/state.json")

def input_fingerprint(paths) -> str:
    h = hashlib.sha256()
    for p in sorted(Path(x) for x in paths):
        stat = p.stat()
        h.update(f"{p.name}:{stat.st_size}:{int(stat.st_mtime)}".encode())
    return h.hexdigest()[:16]

def already_done(step: str, fingerprint: str) -> bool:
    if not STATE.exists():
        return False
    return json.loads(STATE.read_text()).get(step) == fingerprint

def mark_done(step: str, fingerprint: str) -> None:
    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    state[step] = fingerprint
    STATE.parent.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps(state, indent=2))
fingerprint = input_fingerprint(["data/raw/parcels.gpkg", "data/ref/city.gpkg"])
if already_done("clean_parcels", fingerprint):
    print("inputs unchanged — skipping")
else:
    run_clean()
    mark_done("clean_parcels", fingerprint)

This is idempotency turned into a performance feature: a re-run with unchanged inputs becomes a no-op rather than repeated work.

Take the clock as a parameter 

import argparse
from datetime import date

# not reproducible: the result depends on when you ran it
def filter_recent_bad(gdf):
    return gdf[gdf["surveyed_at"] >= date.today().replace(day=1)]

# reproducible: the same run_date always selects the same rows
def filter_recent(gdf, run_date: date):
    return gdf[gdf["surveyed_at"] >= run_date.replace(day=1)]

ap = argparse.ArgumentParser()
ap.add_argument("--run-date", type=date.fromisoformat, default=date.today())
args = ap.parse_args()

A pipeline whose behaviour depends on the wall clock cannot be re-run to reproduce an old result, and cannot be backfilled. Passing the date in costs one argument and buys both.

Notifications and external calls 

import requests

def notify_once(message: str, run_id: str, sent: set) -> None:
    """Alerts are side effects too — do not re-send them on a retry."""
    key = f"{run_id}:{hash(message)}"
    if key in sent:
        return
    requests.post("https://alerts.internal/hook", json={"text": message}, timeout=10)
    sent.add(key)

For an external API that creates something, use an idempotency key if the service supports one — that is exactly what payment APIs mean by the term, and the concept transfers.

Verify the property 

import geopandas as gpd
from pathlib import Path
import hashlib

def file_hash(path: Path) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def test_pipeline_is_idempotent(tmp_path):
    config = {"input": "tests/fixtures/parcels.gpkg",
              "output": str(tmp_path / "out.gpkg"), "run_date": "2026-08-11"}

    first = run(config)
    hash_one = file_hash(config["output"])
    count_one = len(gpd.read_file(config["output"]))

    second = run(config)                       # exactly the same inputs
    assert len(gpd.read_file(config["output"])) == count_one
    assert file_hash(config["output"]) == hash_one, "output changed on re-run"

Row count is the first-order check; a byte-identical hash is the strong one, and it only holds if nothing in the output depends on the current time.

Code examples

Example 1: an idempotent file pipeline 

#!/usr/bin/env python3
from datetime import date
from pathlib import Path
import argparse, hashlib, json
import geopandas as gpd

STATE = Path("logs/state.json")

def fingerprint(*paths) -> str:
    h = hashlib.sha256()
    for path in sorted(Path(p) for p in paths):
        stat = path.stat()
        h.update(f"{path.name}:{stat.st_size}:{int(stat.st_mtime)}".encode())
    return h.hexdigest()[:16]

def write_atomic(gdf, dest: Path) -> Path:
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(dest.name + ".part")
    gdf.to_file(tmp, driver="GPKG")
    tmp.replace(dest)
    return dest

def run(input_path: Path, boundary_path: Path, out_dir: Path,
        run_date: date, force: bool = False) -> dict:
    dest = out_dir / f"parcels_{run_date.isoformat()}.gpkg"
    key = fingerprint(input_path, boundary_path)

    state = json.loads(STATE.read_text()) if STATE.exists() else {}
    if not force and dest.exists() and state.get(str(dest)) == key:
        return {"status": "skipped", "output": str(dest), "reason": "inputs unchanged"}

    gdf = gpd.read_file(input_path)
    boundary = gpd.read_file(boundary_path).to_crs(gdf.crs)
    clipped = gpd.clip(gdf, boundary, keep_geom_type=True)
    clipped = clipped[clipped.geometry.notna() & clipped.geometry.is_valid]

    write_atomic(clipped, dest)
    state[str(dest)] = key
    STATE.parent.mkdir(parents=True, exist_ok=True)
    STATE.write_text(json.dumps(state, indent=2))
    return {"status": "written", "output": str(dest), "features": len(clipped)}

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--run-date", type=date.fromisoformat, default=date.today())
    ap.add_argument("--force", action="store_true")
    args = ap.parse_args()
    print(run(Path("data/raw/parcels.gpkg"), Path("data/ref/city.gpkg"),
              Path("data/out"), args.run_date, args.force))

Run it ten times: one file, one state entry, nine no-ops.

Example 2: an idempotent database load with a run slice 

from sqlalchemy import create_engine, text
import geopandas as gpd

engine = create_engine("postgresql+psycopg://[email protected]/gis")

def replace_slice(gdf, table: str, run_date: str) -> dict:
    with engine.begin() as conn:
        exists = conn.execute(
            text("SELECT to_regclass(:t) IS NOT NULL"), {"t": f"public.{table}"}).scalar()
        deleted = 0
        if exists:
            deleted = conn.execute(
                text(f"DELETE FROM {table} WHERE run_date = :d"), {"d": run_date}).rowcount
        gdf.assign(run_date=run_date).to_postgis(table, conn, if_exists="append", index=False)
    return {"deleted": deleted, "inserted": len(gdf), "run_date": run_date}

print(replace_slice(clean, "parcels", "2026-08-11"))
print(replace_slice(clean, "parcels", "2026-08-11"))    # same end state

Example 3: a resumable batch that is idempotent per file 

from pathlib import Path
import geopandas as gpd

def convert_folder(src: Path, out: Path, force=False) -> dict:
    out.mkdir(parents=True, exist_ok=True)
    written = skipped = failed = 0

    for shp in sorted(src.rglob("*.shp")):
        dest = (out / shp.relative_to(src)).with_suffix(".gpkg")
        if dest.exists() and not force and dest.stat().st_mtime >= shp.stat().st_mtime:
            skipped += 1
            continue
        try:
            dest.parent.mkdir(parents=True, exist_ok=True)
            tmp = dest.with_name(dest.name + ".part")
            gpd.read_file(shp).to_file(tmp, driver="GPKG")
            tmp.replace(dest)
            written += 1
        except Exception as exc:
            print(f"failed {shp.name}: {exc}")
            failed += 1

    return {"written": written, "skipped": skipped, "failed": failed}

Because each file is written atomically and skipped when up to date, an interrupted run can simply be started again — no bookkeeping required beyond the filesystem itself.

Example 4: prove it in the test suite 

import hashlib
from pathlib import Path
import geopandas as gpd

def digest(path) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def test_rerun_produces_identical_output(tmp_path, sample_parcels_file, boundary_file):
    kwargs = dict(input_path=sample_parcels_file, boundary_path=boundary_file,
                  out_dir=tmp_path, run_date=date(2026, 8, 11))

    first = run(**kwargs)
    assert first["status"] == "written"
    first_hash = digest(first["output"])

    second = run(**kwargs)
    assert second["status"] == "skipped"

    third = run(**kwargs, force=True)
    assert digest(third["output"]) == first_hash, "forced re-run changed the bytes"

The third assertion is the strict one: it fails if anything in the output embeds a timestamp or a random value, which is usually a genuine reproducibility bug worth knowing about.

Explanation

Idempotency comes from mathematics — an operation is idempotent when applying it twice gives the same result as applying it once — and in data engineering it means running a job again leaves the system in the same state, not a doubled one.

Vertical steps showing a failed run, a retry, and convergence to a single correct state.
Retry, resume, backfill and reproduce are all the same capability, and they all need this one property.

The reason it matters so much for scheduled work is that failures are normal. A network blip, a locked file, a full disk, a source that published late — none of these are exceptional, and all of them mean the job must run again. If re-running is safe, the response is automatic: the scheduler retries, or someone runs it in the morning, and the state converges. If re-running is not safe, every failure requires a human to determine what state the world is in before doing anything, and that human is usually you, at an inconvenient hour.

The property is broken by three things, and only three. Accumulation: appending rows, adding files, incrementing counters. Non-determinism: output that depends on now(), a random seed, or the order of a directory listing. Partial effects: a job that half-wrote a file or committed some rows and not others, leaving a state neither before nor after.

Each has a standard remedy. Accumulation becomes replacement — overwrite, delete-then-insert within a key, or upsert. Non-determinism becomes parameterisation — take the run date as an argument, seed the generator, sort the file list. Partial effects become atomicity — write to a temporary name and rename, or wrap the database work in a transaction.

The payoff extends well beyond retries. A pipeline that is safe to re-run can be backfilled over historical dates, can skip work whose inputs have not changed, can be tested by running it twice, and can reproduce an old result on demand. Those are four different capabilities, and they all fall out of the same discipline — which is why idempotency is usually the first property worth engineering into a job that runs without you.

Edge cases or notes

  • Path.replace() is atomic on the same filesystem only: Writing the temporary file into a different mount and renaming across it is a copy, not a rename.
  • Appending is sometimes correct: Event logs and audit trails are meant to accumulate. Make that a deliberate exception, keyed so duplicates can be detected.
  • Timestamps inside outputs break byte-identity: A "created_at" column is useful and makes hash comparison impossible. Choose which you want.
  • Deterministic ordering matters: sorted() your file lists, and sort before writing, or two runs produce different byte layouts from the same data.
  • Idempotent is not the same as incremental: Replacing the whole output every run is idempotent and possibly wasteful; fingerprinting inputs adds the efficiency.
  • Watch out for external side effects: Emails, tickets and webhook calls are not undone by a rollback. Guard them with a sent-set or an idempotency key.
  • Partial reruns need a slice key: "This run owns run_date = X" is what makes delete-then-insert safe when several runs share a table.

FAQ

What does idempotent mean for a data job?

Running it twice with the same inputs leaves the same state as running it once. No doubled rows, no extra files, no half-written outputs.

How do I make a database load idempotent?

Delete the slice this run owns and insert it again inside one transaction, or use an upsert keyed on a business identifier. Never plain-append without a key.

Why does using datetime.now() break it?

Because the output then depends on when the job ran rather than on its inputs, so a re-run produces a different file name or a different filtered set. Pass the run date in as a parameter.

What is an atomic write and why does it matter?

Writing to a temporary file and renaming it into place, so the destination never exists in a partial state. Without it, a crash leaves a truncated file that looks finished.

Is skipping unchanged work part of idempotency?

Strictly it is an optimisation on top: idempotency says a re-run is safe, fingerprinting inputs makes it cheap. They pair naturally.

How do I test for it?

Run the pipeline twice in a test and assert the output is unchanged — row count at minimum, byte-identical hash if nothing embeds a timestamp.

Are there cases where appending is right?

Yes — audit logs, event streams and run ledgers are meant to grow. Make it explicit, and key the entries so a repeated run can be detected rather than double-counted.