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. These are the runs further down this page, on HM Land Registry's 21,253 freehold parcels for Rutland:

$ python monthly_load.py                       # to_postgis(..., if_exists="append")
after run 1: 21,253 rows
$ python monthly_load.py                       # after fixing the source file
after run 2: 42,506 rows

Or the file version, three runs of a job that names its output after the clock or numbers it to avoid a clash:

data/out/naming/
  numbered_2026-08-11.gpkg
  numbered_2026-08-11_1.gpkg
  numbered_2026-08-11_1_2.gpkg        # which one is the real one?
  stamped_20260930_101439.gpkg
  stamped_20260930_101441.gpkg
  stamped_20260930_101443.gpkg

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
    tmp = dest.parent / ".partial" / dest.name            # same filesystem, same extension
    tmp.parent.mkdir(parents=True, exist_ok=True)
    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

Replace, write atomically, and make done detectable

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

Three ways of naming the output, each run three times against the same 21,253 Rutland parcels:

import time
from datetime import datetime
from pathlib import Path

out = Path("data/out/naming")
out.mkdir(parents=True, exist_ok=True)

def stamped(gdf):                      # not idempotent — a new file every run
    dest = out / f"stamped_{datetime.now():%Y%m%d_%H%M%S}.gpkg"
    gdf.to_file(dest, driver="GPKG")

def numbered(gdf, run_date):           # not idempotent — versions accumulate
    dest = out / f"numbered_{run_date}.gpkg"
    n = 1
    while dest.exists():
        dest = dest.with_stem(f"{dest.stem}_{n}"); n += 1
    gdf.to_file(dest, driver="GPKG")

def derived(gdf, run_date):            # idempotent — same inputs, same path, replaced
    gdf.to_file(out / f"derived_{run_date}.gpkg", driver="GPKG")

for attempt in range(3):
    stamped(gdf); numbered(gdf, run_date); derived(gdf, run_date)
    time.sleep(1)

for path in sorted(out.iterdir()):
    print(path.name)

derived_2026-08-11.gpkg
numbered_2026-08-11.gpkg
numbered_2026-08-11_1.gpkg
numbered_2026-08-11_1_2.gpkg
stamped_20261011_011526.gpkg
stamped_20261011_011528.gpkg
stamped_20261011_011530.gpkg

Three runs, seven files. The timestamped name gives one file per run and no way to tell which is authoritative, and the numbering loop is worse than it looks: dest.stem already carries the _1 from the previous pass, so the names grow _1_2 rather than counting. Only the derived name leaves one file. 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

To see what a crash leaves behind, each writer below runs in a child process that is killed with SIGKILL 0.4 seconds after it starts writing — about a third of the way through a 21,253-parcel GeoPackage:

import os, signal, subprocess, sys, textwrap
from pathlib import Path
import geopandas as gpd

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

def killed_midway(dest: Path, atomic: bool) -> Path:
    target = dest.parent / ".partial" / dest.name if atomic else dest
    child = textwrap.dedent(f"""
        import geopandas as gpd
        from pathlib import Path
        gdf = gpd.read_file("data/raw/parcels.gpkg")
        dest, target = Path("{dest}"), Path("{target}")
        target.parent.mkdir(parents=True, exist_ok=True)
        gdf.to_file(target, driver="GPKG")
        if target != dest:
            target.replace(dest)
    """)
    proc = subprocess.Popen([sys.executable, "-c", child])
    while not target.exists():             # wait until the write has begun
        time.sleep(0.001)
    time.sleep(0.05)
    proc.send_signal(signal.SIGKILL)
    proc.wait()
    return target

for atomic in (False, True):
    dest = Path(f"data/out/crash/{'atomic' if atomic else 'direct'}.gpkg")
    for leftover in dest.parent.rglob("*"):
        if leftover.is_file():
            leftover.unlink()
    killed_midway(dest, atomic)
    left = [f"{p.relative_to(dest.parent)} ({p.stat().st_size / 1e3:,.0f} kB)"
            for p in sorted(dest.parent.rglob("*")) if p.is_file()]
    try:
        state = f"{len(gpd.read_file(dest)):,} features"
    except Exception as exc:
        state = "read fails: " + str(exc).split(": ")[-1].split(";")[0]
    print(f"{'atomic' if atomic else 'direct'}: left {left}\n        {dest.name}: {state}")
    if not atomic:
        import sqlite3
        with sqlite3.connect(dest) as db:          # a writable open replays the journal
            layers = [row[0] for row in db.execute("SELECT table_name FROM gpkg_contents")]
        print(f"        after a writable open: {dest.stat().st_size / 1e3:,.0f} kB, layers {layers}")

direct: left ['direct.gpkg (5,874 kB)', 'direct.gpkg-journal (46 kB)']
        direct.gpkg: read fails: attempt to write a readonly database
        after a writable open: 66 kB, layers []
atomic: left ['.partial/atomic.gpkg (5,612 kB)', '.partial/atomic.gpkg-journal (46 kB)']
        atomic.gpkg: read fails: No such file or directory

The direct write leaves a file at the real name — the thing every "does the output exist?" check looks for — and it is not the dataset. A GeoPackage is a SQLite database, so what is left is a half-written file plus the rollback journal SQLite uses to undo an unfinished transaction. A read-only open refuses it. Anything that opens it for writing replays the journal, and what that recovers is a 66 kB GeoPackage with no layer in it at all. The atomic version leaves its debris in .partial/, which no downstream step reads and the next run overwrites, and the real name does not exist until the rename. The destination only ever appears complete.

Databases: delete-then-insert inside a transaction

First, the load from the top of the page: an append run twice.

from sqlalchemy import text

def count(table: str, where: str = "true") -> int:
    with engine.connect() as conn:
        return conn.execute(text(f"SELECT count(*) FROM {table} WHERE {where}")).scalar()

with engine.begin() as conn:
    conn.execute(text("DROP TABLE IF EXISTS parcels_naive, parcels_slices, parcels_upsert"))

for attempt in range(2):
    gdf.to_postgis("parcels_naive", engine, if_exists="append", index=False)
    print(f"after run {attempt + 1}: {count('parcels_naive'):,} rows")

after run 1: 21,253 rows
after run 2: 42,506 rows

The fix is to make each run own a slice of the table and replace it:

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
        if conn.execute(text("SELECT to_regclass(:t)"), {"t": table}).scalar():
            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)
        return conn.execute(
            text(f"SELECT count(*) FROM {table} WHERE run_date = :d"), {"d": run_date}
        ).scalar()

for attempt in range(3):
    print(f"run {attempt + 1}: slice {load_partition(gdf, engine, 'parcels_slices', run_date):,}"
          f"  table {count('parcels_slices'):,}")
load_partition(gdf, engine, "parcels_slices", "2026-09-11")
print(f"after a second run date: table {count('parcels_slices'):,}")

run 1: slice 21,253  table 21,253
run 2: slice 21,253  table 21,253
run 3: slice 21,253  table 21,253
after a second run date: table 42,506

The pattern is delete the slice this run owns, then insert it, both inside one transaction. The to_regclass check matters: without it the very first run — the one with no table yet — fails on the DELETE with UndefinedTable. Passing the open connection to to_postgis keeps the insert in the same transaction as the delete, which is what makes a failure roll back to the previous state (Example 2 shows it). if_exists="replace" also works when the run owns the whole table.

An upsert achieves the same at row level, keyed on the Land Registry's INSPIREID:

import shapely

with engine.begin() as conn:
    conn.execute(text("""CREATE TABLE parcels_upsert (
        parcel_id bigint PRIMARY KEY, valid_from timestamptz,
        geom geometry(Polygon, 27700), updated_at date)"""))

rows = [{"parcel_id": int(r.INSPIREID), "valid_from": r.VALIDFROM,
         "geom": shapely.to_wkb(r.geometry), "updated_at": run_date}
        for r in gdf.itertuples()]

upsert = text("""
INSERT INTO parcels_upsert (parcel_id, valid_from, geom, updated_at)
VALUES (:parcel_id, :valid_from, ST_GeomFromWKB(:geom, 27700), :updated_at)
ON CONFLICT (parcel_id) DO UPDATE
SET valid_from = EXCLUDED.valid_from, geom = EXCLUDED.geom, updated_at = EXCLUDED.updated_at""")

for attempt in range(2):
    start = time.perf_counter()
    with engine.begin() as conn:
        conn.execute(upsert, rows)
    print(f"run {attempt + 1}: {count('parcels_upsert'):,} rows  ({time.perf_counter() - start:.1f} s)")

run 1: 21,253 rows  (0.3 s)
run 2: 21,253 rows  (0.3 s)

Make "already done" detectable

import hashlib, json, os
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))

def run_clean():
    parcels = gpd.read_file("data/raw/parcels.gpkg")
    ward = gpd.read_file("data/ref/oakham.gpkg").to_crs(parcels.crs)
    gpd.clip(parcels, ward).to_file("data/out/clean_parcels.gpkg")

inputs = ["data/raw/parcels.gpkg", "data/ref/oakham.gpkg"]

def step():
    fingerprint = input_fingerprint(inputs)
    if already_done("clean_parcels", fingerprint):
        return "inputs unchanged — skipping"
    run_clean()
    mark_done("clean_parcels", fingerprint)
    return "ran"

print("first run: ", step())
print("second run:", step())
os.utime(inputs[0])                      # same bytes, new mtime: what a re-download does
print("after touch:", step())

first run:  ran
second run: inputs unchanged — skipping
after touch: ran

The metadata fingerprint is cheap and conservative: it re-runs whenever the file might have changed, and a re-download of byte-identical data counts as a change. Hashing the contents instead costs a read of every input:

def content_fingerprint(paths) -> str:
    h = hashlib.sha256()
    for p in sorted(Path(x) for x in paths):
        with open(p, "rb") as f:
            for block in iter(lambda: f.read(1 << 20), b""):
                h.update(block)
    return h.hexdigest()[:16]

size = sum(Path(p).stat().st_size for p in inputs)
start = time.perf_counter()
before = content_fingerprint(inputs)
elapsed = time.perf_counter() - start
os.utime(inputs[0])
print(f"hashed {size / 1e6:.1f} MB in {elapsed * 1000:.0f} ms; unchanged after touch: "
      f"{content_fingerprint(inputs) == before}")

hashed 21.1 MB in 12 ms; unchanged after touch: True

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

The Land Registry records when each parcel's current version began (BEGINLIFESPANVERSION). A job that selects "parcels changed this month":

import argparse
from datetime import date
import pandas as pd

def month_start(day: date) -> pd.Timestamp:
    return pd.Timestamp(day.replace(day=1), tz="UTC")

# not reproducible: the result depends on when you ran it
def changed_this_month_bad(gdf):
    return gdf[gdf["BEGINLIFESPANVERSION"] >= month_start(date.today())]

# reproducible: the same run_date always selects the same rows
def changed_since(gdf, run_date: date):
    return gdf[gdf["BEGINLIFESPANVERSION"] >= month_start(run_date)]

ap = argparse.ArgumentParser()
ap.add_argument("--run-date", type=date.fromisoformat, default=date.today())
args = ap.parse_args(["--run-date", "2023-03-15"])     # a backfill for March 2023

print("newest version in the file:", gdf["BEGINLIFESPANVERSION"].max().date())
print("wall clock:  ", len(changed_this_month_bad(gdf)), "parcels")
print("--run-date:  ", len(changed_since(gdf, args.run_date)), "parcels")

newest version in the file: 2023-03-02
wall clock:   0 parcels
--run-date:   5 parcels

The wall-clock version returns nothing, and would have returned something different in March 2023 — it cannot reproduce its own history. 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

A local stand-in for the alert webhook counts what arrives. The first version keeps its "already sent" set in memory and keys it on hash(message):

import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import requests

received = []
class Hook(BaseHTTPRequestHandler):
    def do_POST(self):
        received.append(self.rfile.read(int(self.headers["Content-Length"])))
        self.send_response(204); self.end_headers()
    def log_message(self, *args):
        pass

server = HTTPServer(("127.0.0.1", 0), Hook)
threading.Thread(target=server.serve_forever, daemon=True).start()
HOOK = f"http://127.0.0.1:{server.server_port}/hook"

def notify_once(message: str, run_id: str, sent: set) -> None:
    key = f"{run_id}:{hash(message)}"
    if key in sent:
        return
    requests.post(HOOK, json={"text": message}, timeout=10)
    sent.add(key)

message = "rutland parcels: 21,253 loaded"
for attempt in range(2):          # a retry is a new process, so it starts with an empty set
    notify_once(message, "2026-08-11", set())
print("alerts received:", len(received))

env = {k: v for k, v in os.environ.items() if k != "PYTHONHASHSEED"}   # a pinned seed would hide it
keys = {subprocess.run([sys.executable, "-c", f"print(hash({message!r}))"],
                       capture_output=True, text=True, env=env).stdout.strip() for _ in range(3)}
print("distinct hash() values in three processes:", len(keys))

alerts received: 2
distinct hash() values in three processes: 3

Both halves fail: a set in memory does not survive into the retry, and hash() of a string is salted per process, so even a persisted key would not match. The version that works stores a stable digest on disk:

SENT = Path("logs/sent.txt")

def notify_once(message: str, run_id: str) -> bool:
    """Alerts are side effects too — do not re-send them on a retry."""
    key = f"{run_id}:{hashlib.sha256(message.encode()).hexdigest()[:16]}"
    if SENT.exists() and key in SENT.read_text().split():
        return False
    requests.post(HOOK, json={"text": message}, timeout=10)
    SENT.parent.mkdir(parents=True, exist_ok=True)
    with SENT.open("a") as f:
        f.write(key + "\n")
    return True

received.clear()
print([notify_once(message, "2026-08-11") for attempt in range(3)], "received:", len(received))
server.shutdown()

[True, False, False] received: 1

The key is written after the post succeeds, so a crash between the two re-sends one alert rather than losing it — for an alert that is the right way round. 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

The test is to run the job twice and compare. Here it is on the load() from the quick answer:

import sqlite3
import pyogrio

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

def last_change(path):
    with sqlite3.connect(path) as db:
        return db.execute("SELECT last_change FROM gpkg_contents").fetchone()[0]

first = load(gdf, run_date); hash_one, stamp_one = file_hash(first), last_change(first)
time.sleep(1)
second = load(gdf, run_date)
print("rows equal:", len(gpd.read_file(second)) == len(gdf))
print("bytes equal:", file_hash(second) == hash_one, "|", stamp_one, "->", last_change(second))

pyogrio.set_gdal_config_options({"OGR_CURRENT_DATE": f"{run_date}T00:00:00Z"})
hashes = {file_hash(load(gdf, run_date)) for attempt in range(2)}
print("with OGR_CURRENT_DATE:", len(hashes), "distinct hash")
pyogrio.set_gdal_config_options({"OGR_CURRENT_DATE": None})

rows equal: True
bytes equal: False | 2026-10-10T23:15:40.694Z -> 2026-10-10T23:15:41.927Z
with OGR_CURRENT_DATE: 1 distinct hash

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. GeoPackage is itself such an output: GDAL stamps gpkg_contents.last_change with the time of writing, so two otherwise identical runs differ in those bytes. Setting OGR_CURRENT_DATE to the run date makes the stamp a function of the inputs like everything else.

Re-running the same Rutland job until it is safe

The data used on this page

HM Land Registry's INSPIRE Index Polygons for Rutland (21,253 freehold parcels), the ONS boundary of the Oakham North West ward, Geofabrik's Rutland OpenStreetMap shapefiles for the batch in Example 3, and a throwaway PostGIS 3.4 in Docker for the database loads. gdf and run_date are used throughout.

import io
import os
import time
import zipfile
from pathlib import Path

import geopandas as gpd
import pyogrio
import requests
from sqlalchemy import create_engine

pyogrio.set_gdal_config_options({"GML_DOWNLOAD_SCHEMA": "NO"})
for folder in ["data/raw", "data/ref", "data/shp"]:
    Path(folder).mkdir(parents=True, exist_ok=True)

if not Path("data/raw/parcels.gpkg").exists():
    reply = requests.get("https://use-land-property-data.service.gov.uk/datasets/inspire/download/"
                         "Rutland.zip", timeout=600)
    reply.raise_for_status()
    zipfile.ZipFile(io.BytesIO(reply.content)).extract("Land_Registry_Cadastral_Parcels.gml", "downloads")
    gpd.read_file("downloads/Land_Registry_Cadastral_Parcels.gml").to_file("data/raw/parcels.gpkg")
if not Path("data/ref/oakham.gpkg").exists():
    reply = requests.get("https://services1.arcgis.com/ESMARspQHYMw9BZ9/arcgis/rest/services/"
                         "Wards_December_2024_Boundaries_UK_BGC/FeatureServer/0/query",
                         params={"where": "WD24NM='Oakham North West'", "outFields": "WD24CD,WD24NM",
                                 "outSR": 4326, "f": "geojson"}, timeout=120)
    reply.raise_for_status()
    gpd.read_file(io.BytesIO(reply.content)).to_file("data/ref/oakham.gpkg")
if not Path("downloads/rutland-shp.zip").exists():
    reply = requests.get("https://download.geofabrik.de/europe/united-kingdom/england/"
                         "rutland-260101-free.shp.zip", timeout=300)
    reply.raise_for_status()
    Path("downloads/rutland-shp.zip").write_bytes(reply.content)
zipfile.ZipFile("downloads/rutland-shp.zip").extractall("data/shp")

os.environ["PGPASSWORD"] = "guide"
if os.system("docker inspect guide-postgis >/dev/null 2>&1") != 0:
    os.system("docker run -d --name guide-postgis -e POSTGRES_USER=user -e POSTGRES_PASSWORD=guide "
              "-e POSTGRES_DB=gis -p 127.0.0.1:55433:5432 postgis/postgis:16-3.4 >/dev/null")
    time.sleep(10)
engine = create_engine("postgresql+psycopg://user@localhost:55433/gis")

gdf = gpd.read_file("data/raw/parcels.gpkg")
run_date = "2026-08-11"
print(f"{len(gdf):,} parcels, {gdf.crs}; "
      f"{len(list(Path('data/shp').glob('*.shp')))} shapefiles; "
      f"ward: {gpd.read_file('data/ref/oakham.gpkg').WD24NM[0]}")

21,253 parcels, EPSG:27700; 18 shapefiles; ward: Oakham North West

Example 1: an idempotent file pipeline

%%writefile idem_pipeline.py
#!/usr/bin/env python3
from datetime import date
from pathlib import Path
import argparse, hashlib, json
import geopandas as gpd
import pyogrio

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, run_date: date) -> Path:
    tmp = dest.parent / ".partial" / dest.name
    tmp.parent.mkdir(parents=True, exist_ok=True)
    pyogrio.set_gdal_config_options({"OGR_CURRENT_DATE": f"{run_date.isoformat()}T00:00:00Z"})
    try:
        gdf.to_file(tmp, driver="GPKG")
    finally:
        pyogrio.set_gdal_config_options({"OGR_CURRENT_DATE": None})
    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, run_date)
    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)}

def main(argv=None):
    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(argv)
    print(run(Path("data/raw/parcels.gpkg"), Path("data/ref/oakham.gpkg"),
              Path("data/out/pipeline"), args.run_date, args.force))

if __name__ == "__main__":
    main()

Run it ten times from the shell, as a scheduler would:

import ast, collections
runs = [subprocess.run([sys.executable, "idem_pipeline.py", "--run-date", "2026-08-11"],
                       capture_output=True, text=True).stdout.strip() for attempt in range(10)]
print(runs[0])
print(runs[1])
print(collections.Counter(ast.literal_eval(r)["status"] for r in runs))
print(sorted(p.name for p in Path("data/out/pipeline").iterdir()), "|", Path("logs/state.json").read_text().count(".gpkg"), "state entry")

{'status': 'written', 'output': 'data/out/pipeline/parcels_2026-08-11.gpkg', 'features': 1425}
{'status': 'skipped', 'output': 'data/out/pipeline/parcels_2026-08-11.gpkg', 'reason': 'inputs unchanged'}
Counter({'skipped': 9, 'written': 1})
['.partial', 'parcels_2026-08-11.gpkg'] | 1 state entry

One file (beside the empty .partial/ folder its writes pass through), one state entry, nine no-ops. The clip keeps the 1,425 parcels that fall in the Oakham North West ward.

Example 2: a failed load leaves the previous state

The point of the transaction is what happens when the insert fails after the delete. Here the second load is handed a frame with a column the table does not have:

def replace_slice(gdf, table: str, run_date: str) -> dict:
    with engine.begin() as conn:
        exists = conn.execute(text("SELECT to_regclass(:t)"), {"t": 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}

this_slice = "run_date = '2026-08-11'"
print(replace_slice(gdf, "parcels_slices", "2026-08-11"))
print(f"slice now: {count('parcels_slices', this_slice):,}")
try:
    replace_slice(gdf.assign(surveyor="unknown"), "parcels_slices", "2026-08-11")
except Exception as exc:
    print("failed:", type(exc).__name__, str(exc).splitlines()[0])
print(f"slice after the failure: {count('parcels_slices', this_slice):,}")

{'deleted': 21253, 'inserted': 21253, 'run_date': '2026-08-11'}
slice now: 21,253
failed: UndefinedColumn column "surveyor" of relation "parcels_slices" does not exist
slice after the failure: 21,253

The DELETE ran — and was rolled back with the failed insert, so the table still holds the last good load. Had the delete been committed on its own, the failure would have left the slice empty, which is worse than doubled.

Example 3: a resumable batch that is idempotent per file

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:
            tmp = dest.parent / ".partial" / dest.name
            tmp.parent.mkdir(parents=True, exist_ok=True)
            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}

for label in ["first run", "second run"]:
    start = time.perf_counter()
    print(f"{label:22}", convert_folder(Path("data/shp"), Path("data/gpkg")),
          f"{time.perf_counter() - start:.1f} s")

zipfile.ZipFile("downloads/rutland-shp.zip").extractall("data/shp")   # the same download, unpacked again
start = time.perf_counter()
print(f"{'after re-extracting':22}", convert_folder(Path("data/shp"), Path("data/gpkg")),
      f"{time.perf_counter() - start:.1f} s")

first run              {'written': 18, 'skipped': 0, 'failed': 0} 0.2 s
second run             {'written': 0, 'skipped': 18, 'failed': 0} 0.0 s
after re-extracting    {'written': 18, 'skipped': 0, 'failed': 0} 0.2 s

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. The price of using modification times is visible in the third line: unpacking the same archive again gives every shapefile a new mtime, and the whole folder is converted again. That is wasted work, never a wrong result, which is the right way for a skip test to err.

Example 4: prove it in the test suite

%%writefile test_idem_pipeline.py
import hashlib
from datetime import date
from pathlib import Path

import pytest
from idem_pipeline import run

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

@pytest.fixture
def kwargs(tmp_path):
    return dict(input_path=Path("data/raw/parcels.gpkg"), boundary_path=Path("data/ref/oakham.gpkg"),
                out_dir=tmp_path, run_date=date(2026, 8, 11))

def test_rerun_produces_identical_output(kwargs):
    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"

!{sys.executable} -m pytest -q -p no:cacheprovider test_idem_pipeline.py 2>&1 | tail -2

.                                                                        [100%]
1 passed in 0.59s

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. It passes here only because write_atomic pins OGR_CURRENT_DATE; without that line the same test fails on the GeoPackage's own last_change stamp, as the section above measured.

Same result twice as once

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.

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.

Renames across filesystems and other atomicity gaps

• 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 — and Path.replace() refuses it outright with OSError: [Errno 18] Invalid cross-device link.
• Keep the extension on the temporary file: Writing a GeoPackage to parcels.gpkg.part works, but GDAL warns on every write that the extension should be gpkg. A .partial/ folder beside the output keeps the real name and the real extension, and a *.gpkg glob of the output folder never picks it up.
• 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.

Internal links

• How to Build a Resumable Batch GIS Job in Python
• How to Cache Pipeline Steps So Only Changed Data Is Reprocessed
• How to Make a GIS Workflow Reproducible in Python
• What Is a GIS Data Pipeline? The Anatomy Explained
• Batch Output Files Keep Overwriting Each Other: How to Fix It
• How to Add Retries and Timeouts to an Automated GIS Job in Python

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 file at the real name that is not the dataset: a GeoPackage killed mid-write here could not be opened read-only, and recovered to an empty 66 kB file.

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. A GeoPackage does (gpkg_contents.last_change) unless OGR_CURRENT_DATE is set.

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.