How to record lineage automatically in a pipeline

Problem statement

Lineage that a person writes is lineage that stops being written the week the deadline moves. The only records that stay accurate are the ones the pipeline produces as a side effect of doing the work, from the same variables that did it.

That means three things have to happen in the code rather than in a document: inputs get registered when they are opened, steps get registered when they run, and the record gets written in the same function that writes the output. Everything else — the environment, the code version, the timestamps — is free.

This guide builds that as a small context manager, wires it into a pipeline, and makes the resulting store queryable.

Quick answer

import json, pathlib, datetime, subprocess, sys, hashlib
import importlib.metadata as md

class Run:
    def __init__(self, output, store="lineage"):
        self.output, self.store = str(output), pathlib.Path(store)
        self.rec = {"output": str(output), "inputs": [], "steps": [],
                    "started": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")}

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, tb):
        self.rec["status"] = "failed" if exc else "ok"
        if exc:
            self.rec["error"] = f"{exc_type.__name__}: {exc}"
        self.rec["finished"] = datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
        self.rec["environment"] = environment()
        if not exc and pathlib.Path(self.output).exists():
            self.rec["output_sha256"] = file_hash(self.output)
        self.store.mkdir(parents=True, exist_ok=True)
        name = pathlib.Path(self.output).name + f".{self.rec['started'].replace(':', '')}.json"
        (self.store / name).write_text(json.dumps(self.rec, indent=2))
        return False                                  # never swallow the exception

    def input(self, name, path, **kw):
        self.rec["inputs"].append({"name": name, "path": str(path),
                                   "sha256": file_hash(path), **kw})

    def step(self, op, **params):
        self.rec["steps"].append({"op": op, "params": params,
                                  "at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")})

The two design decisions that make it work: it records failures as well as successes, and it never swallows the exception.

Flow from opening inputs, through recorded steps, to a written output and a lineage record. — The record is a by-product of the run, produced from the same variables.

Step-by-step solution

1. Scope the record to one output

One record per artefact, named after it. That is the unit people ask about.

2. Register inputs where they are opened

Wrap the reader, so an input cannot enter the pipeline unrecorded:

def read_vector(run, name, path, **kw):
    run.input(name, path, **kw)
    return gpd.read_file(path)

3. Hash inputs as they are on disk

For files you received, the byte hash is the right thing and is cheap. For large inputs, hash once and cache by path plus mtime plus size.

4. Record parameters that change the answer

The threshold, the resampling method, the CRS, the tolerance, the seed. Not the chunk size or the worker count.

5. Capture the environment once, at the end

Python version, the library versions that do the arithmetic, the git commit and whether the tree was dirty.

6. Record failures

A run that raised still explains a gap in a series, and a record with status: failed and the error is the fastest way to answer "why is there no output for the 14th?".

7. Write to a store, not only beside the output

A directory of records lets you ask questions across runs — which outputs used a source that has turned out to be wrong, which runs used the library version that had the bug.

8. Put the record in the delivery

A copy beside the artefact travels with it; the store stays behind.

Three panels showing the three questions a lineage store answers: which inputs, which outputs, which version. — The store earns its place the first time somebody reports a bad input.

Code examples

Example 1 — the helpers

import hashlib, subprocess, sys, pathlib
import importlib.metadata as md

def file_hash(path, chunk=1 << 20):
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for block in iter(lambda: fh.read(chunk), b""):
            h.update(block)
    return h.hexdigest()

def environment(packages=("geopandas", "shapely", "pyproj", "rasterio", "pandas", "numpy")):
    def version(p):
        try:
            return md.version(p)
        except md.PackageNotFoundError:
            return None
    try:
        commit = subprocess.check_output(["git", "rev-parse", "--short", "HEAD"],
                                         text=True, stderr=subprocess.DEVNULL).strip()
        dirty = bool(subprocess.check_output(["git", "status", "--porcelain"],
                                             text=True, stderr=subprocess.DEVNULL).strip())
    except Exception:
        commit, dirty = None, None
    return {"python": sys.version.split()[0],
            "packages": {p: version(p) for p in packages},
            "git": {"commit": commit, "dirty": dirty}}

Example 2 — a pipeline that records itself

import geopandas as gpd

with Run("flood_zones_2026.gpkg") as run:
    dtm = read_raster(run, "ea_lidar_dtm", "dtm_2024.tif",
                      version="2024-06", licence="OGL-UK-3.0")
    model = read_raster(run, "ea_fluvial_model", "rp200.tif", version="v7")

    run.step("breach_depressions", library="pysheds")
    filled = breach(dtm)

    run.step("flowdir", routing="d8")
    direction = flowdir(filled, routing="d8")

    run.step("threshold", return_period_years=200)
    mask = model > 0

    run.step("polygonise", min_area_m2=100, simplify_m=1.0)
    zones = polygonise(mask, min_area=100).simplify(1.0)

    run.step("write", driver="GPKG", layer="flood_zones")
    zones.to_file("flood_zones_2026.gpkg", layer="flood_zones", driver="GPKG")

If polygonise raises, the record still lands, with the four steps that ran, the two inputs and the traceback's first line.

Example 3 — query the store

import json, pathlib, collections

def load_store(store="lineage"):
    return [json.loads(p.read_text()) for p in pathlib.Path(store).glob("*.json")]

def outputs_using(name_or_hash, store="lineage"):
    hits = []
    for rec in load_store(store):
        for inp in rec["inputs"]:
            if name_or_hash in (inp["name"], inp.get("sha256")):
                hits.append((rec["output"], rec["started"], rec.get("status")))
    return sorted(hits, key=lambda h: h[1], reverse=True)

def library_versions_used(store="lineage"):
    seen = collections.defaultdict(set)
    for rec in load_store(store):
        for pkg, ver in (rec.get("environment", {}).get("packages") or {}).items():
            seen[pkg].add(ver)
    return {k: sorted(v) for k, v in seen.items()}

print(outputs_using("ea_lidar_dtm")[:5])
print(library_versions_used())

library_versions_used is the query nobody expects to need and everybody eventually does, when a result changes and nothing in the code did.

Explanation

Why a context manager rather than a decorator

The record has to be written whether the body succeeded or failed, and it has to see the output file after it was written. __exit__ runs in both cases and runs last, which is exactly the shape of the problem. A decorator on the pipeline function would work too; a decorator per step would not, because the steps do not know about the output.

Why recording failures matters more than it sounds

Most questions about a pipeline are about something that is missing, not something that is wrong. A store that contains only successes cannot answer them, and the person asking has to reconstruct what happened from logs that have rotated.

Why the input hash is the field that pays off

Paths lie: the file at /data/current/dtm.tif is not the file that was there in January. A content hash turns "we used the 2024 DTM" into a claim you can test against the file you still have, and it is what distinguishes a data change from a code change when a result moves.

Why the store needs to stay small enough to read

A JSON file per run is fine for thousands of runs and unmanageable for millions. When it grows, load the records into DuckDB or SQLite and query them there — the schema is stable and shallow, so it maps onto a table without much effort.

Two panels separating lineage parameters worth recording — thresholds, methods, CRS, seeds — from those that only affect speed. — Recording everything reliably is impossible; recording what changes the answer is not.

Edge cases or notes

Hash large inputs once. Cache by path, size and mtime.
Do not record credentials. Record the endpoint and the account, not the token.
Record the URL and the fetch time for downloads. Remote data changes silently.
A dirty git tree invalidates the commit. Record the flag and mean it.
Manual steps are lineage too. Record "edited by hand in QGIS" rather than omitting it.
Timestamps in UTC. Local times across a DST boundary are unsortable.
Rotate or index the store. A directory listing is not a query plan.
Ship a copy with the artefact. The store does not travel.

Internal links

Provenance and lineage explained for spatial workflows — what the record has to contain
How to record run metadata and lineage in a GIS pipeline — the pipeline-side pattern
How to checksum spatial datasets so you can prove they match — hashing outputs rather than inputs
You cannot tell which data version produced a map — the failure this prevents
Reproducible GIS environments explained — pinning what the record names
How to validate pipeline inputs and outputs in Python — the checks that run alongside
Observability for GIS pipelines — where lineage sits among logs and metrics
Dataset versioning explained — the version the record refers to

FAQ

How do I record lineage without adding work to every script?

Use a context manager that captures the environment and writes the record on exit, plus wrapped readers that register inputs as they are opened. The only manual calls left are one step per decision.

Should I record failed runs?

Yes. Most questions are about a missing output, and a record with a status and an error answers them immediately.

What parameters belong in the record?

Ones that change the result: thresholds, resampling methods, tolerances, CRSs, seeds. Not chunk sizes or worker counts.

Where should records be stored?

Both beside the artefact, so they travel, and in a central store, so you can ask which outputs used a given input.

How do I find every output that used a bad source?

Query the store for records whose inputs include that name or content hash. That query is the main reason to keep records centrally.

Do I need to hash every input?

Every input you rely on. Hash large files once and cache the digest by path, size and modification time.

How to record lineage automatically in a pipeline #

Problem statement #

Quick answer #

Step-by-step solution #

1. Scope the record to one output #

2. Register inputs where they are opened #

3. Hash inputs as they are on disk #

4. Record parameters that change the answer #

5. Capture the environment once, at the end #

6. Record failures #

7. Write to a store, not only beside the output #

8. Put the record in the delivery #

Code examples #

Example 1 — the helpers #

Example 2 — a pipeline that records itself #

Example 3 — query the store #

Explanation #

Why a context manager rather than a decorator #

Why recording failures matters more than it sounds #

Why the input hash is the field that pays off #

Why the store needs to stay small enough to read #

Edge cases or notes #

Internal links #

FAQ #

How do I record lineage without adding work to every script? #

Should I record failed runs? #

What parameters belong in the record? #

Where should records be stored? #

How do I find every output that used a bad source? #

Do I need to hash every input? #