# Capturing Processing Lineage with PROV-O in Python

To capture a geoprocessing step as machine-readable lineage, model the input and output datasets as prov:Entity nodes, the operation as a prov:Activity, and connect them with prov:used, prov:wasGeneratedBy, and prov:wasDerivedFrom triples using rdflib, then serialize to Turtle and JSON-LD.

This guide is the runnable companion to Spatial Data Lineage & Provenance Tracking, which covers the full lineage model and its ISO 19115 projection, and it fits inside the wider Spatial Data Audit Reporting & Compliance Governance program that governs how provenance is retained and reported. Here we focus on one thing: a single, self-contained script you can drop into a pipeline to record one processing step correctly.

# Automated Python Implementation

The script below records a reprojection step. It hashes the input and output files to build stable entity URIs, opens a prov:Activity around the work, and writes the derivation triangle. It uses only rdflib, which bundles a ready-bound PROV namespace, so nothing here depends on network access or a specific catalog.

#!/usr/bin/env python3
"""Capture a single geoprocessing step as a PROV-O graph.

Usage:
    python capture_lineage.py input.gpkg output.gpkg \
        --activity reproject --out provenance/step

Produces provenance/step.ttl and provenance/step.jsonld.
"""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
from pathlib import Path

from rdflib import Graph, Literal, URIRef
from rdflib.namespace import PROV, RDFS, XSD

BASE = "https://data.example.gov/prov/"

def content_hash(path: Path) -> str:
    """Return a short SHA-256 digest of a file's contents."""
    digest = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()[:32]

def entity(graph: Graph, path: Path, role: str) -> URIRef:
    """Register a dataset version as a prov:Entity and return its URI."""
    uri = URIRef(f"{BASE}entity/{content_hash(path)}")
    graph.add((uri, RDFS.label, Literal(f"{role}: {path.name}")))
    graph.add((uri, PROV.atLocation, Literal(str(path))))
    return uri

def build_graph(
    in_path: Path,
    out_path: Path,
    activity_name: str,
    agent_label: str = "geoprocessing-pipeline v2.4.1",
) -> Graph:
    """Build a PROV-O graph for one processing step."""
    g = Graph()
    g.bind("prov", PROV)
    g.bind("rdfs", RDFS)

    source = entity(g, in_path, "source")
    derived = entity(g, out_path, "derived")

    activity = URIRef(f"{BASE}activity/{content_hash(out_path)}")
    agent = URIRef(f"{BASE}agent/pipeline")

    now = dt.datetime.now(dt.timezone.utc)
    g.add((activity, RDFS.label, Literal(activity_name)))
    g.add((activity, PROV.startedAtTime, Literal(now, datatype=XSD.dateTime)))
    g.add((activity, PROV.endedAtTime, Literal(now, datatype=XSD.dateTime)))

    g.add((agent, RDFS.label, Literal(agent_label)))
    g.add((agent, PROV.type, PROV.SoftwareAgent))

    # The derivation triangle
    g.add((activity, PROV.used, source))
    g.add((activity, PROV.wasAssociatedWith, agent))
    g.add((derived, PROV.wasGeneratedBy, activity))
    g.add((derived, PROV.wasDerivedFrom, source))
    g.add((derived, PROV.wasAttributedTo, agent))

    return g

def main() -> None:
    parser = argparse.ArgumentParser(description="Capture geoprocessing lineage as PROV-O.")
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument("--activity", default="reproject")
    parser.add_argument("--out", type=Path, default=Path("provenance/step"))
    args = parser.parse_args()

    args.out.parent.mkdir(parents=True, exist_ok=True)
    graph = build_graph(args.input, args.output, args.activity)
    graph.serialize(destination=f"{args.out}.ttl", format="turtle")
    graph.serialize(destination=f"{args.out}.jsonld", format="json-ld", indent=2)
    print(f"Wrote {len(graph)} triples to {args.out}.ttl and {args.out}.jsonld")

if __name__ == "__main__":
    main()

The one line worth flagging is the agent typing: g.add((agent, PROV.type, PROV.SoftwareAgent)) declares that the responsible party is software rather than a person, which matters when an auditor needs to distinguish an automated transformation from a manual edit. Every other triple is exactly what a PROV consumer expects.

# Validation and pipeline integration

Run the script, then confirm the graph parses and contains the derivation core:

python capture_lineage.py input.gpkg output.gpkg \
    --activity reproject --out provenance/step

# Confirm the Turtle round-trips and count triples
python -c "import rdflib; g=rdflib.Graph().parse('provenance/step.ttl'); print(len(g), 'triples')"

A short pytest locks in the invariant that a derived entity is always linked back to its source:

from pathlib import Path
from rdflib.namespace import PROV
from capture_lineage import build_graph

def test_derivation_is_present(tmp_path):
    src = tmp_path / "in.gpkg"
    out = tmp_path / "out.gpkg"
    src.write_bytes(b"source-bytes")
    out.write_bytes(b"derived-bytes")
    g = build_graph(src, out, "reproject")
    derivations = list(g.subject_objects(PROV.wasDerivedFrom))
    assert len(derivations) == 1
    assert list(g.objects(None, PROV.wasGeneratedBy)), "missing generating activity"

To gate it in CI, add the check to a pull-request workflow so any pipeline change that stops emitting lineage is caught before merge — the same enforcement pattern used across CI/CD validation and policy enforcement for spatial data:

# .github/workflows/lineage-smoke.yml
name: Lineage smoke test
on:
  pull_request:
    paths:
      - 'pipelines/**/*.py'
      - 'capture_lineage.py'
jobs:
  lineage:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install "rdflib>=7.0" pytest
      - run: pytest tests/test_capture_lineage.py

# Long-term compliance best practices

• Key entities by content, not path. A SHA-256 of the file contents makes the same dataset version resolve to the same node across runs, machines, and years, so independently captured steps merge into one coherent chain.
• Always type timestamps as xsd:dateTime. Pass datatype=XSD.dateTime and an ISO 8601 string with a timezone, or SPARQL date-range queries over your provenance silently return nothing.
• Record parameters, not just the verb. “reproject” alone is not auditable; “reproject from EPSG:27700 to EPSG:4326” is. Attach the operative parameters to the activity.
• Pin and record the agent version. The software that ran the step is part of the evidence. Bump the agent label whenever the pipeline changes so a defect can be scoped to the runs it affected.
• Serialize to both Turtle and JSON-LD. Turtle is the human-auditable archival copy; JSON-LD feeds the same harvesters that consume your catalog records, keeping one provenance model for two audiences.
• Hash the serialized graph into your evidence store so the lineage record itself cannot be edited after the fact.

# Identity Is the Hard Part

The graph construction in the script above is mechanical. The part that determines whether the resulting provenance is useful — and the part that quietly goes wrong — is deciding what a node’s identifier is.

An Entity is a specific state of a dataset, not the dataset. This is the distinction that most implementations blur. roads is not an Entity; roads as it stood after the 2026-03 import, with content hash abc123 is. Using a stable name as the Entity identifier produces a graph in which a layer was derived from itself, repeatedly, which is both true and useless. Content-addressing the Entity — identifier derived from the hash — makes derivation edges meaningful and makes the graph append-only by construction.

An Activity needs an identity that survives a re-run. If the same clip is run twice with identical inputs and parameters, is that one Activity or two? Both answers are defensible and the choice must be made explicitly. Minting a fresh identifier per execution records operational history and grows the graph steadily; deriving the identifier from the inputs and parameters makes re-runs idempotent and loses the fact that it happened twice. For compliance purposes the second is usually right, with the execution history kept in the audit trail where repetition is the point.

An Agent should be a role plus a version, not a person. People leave and their identifiers stop resolving; teams and software versions persist. planning-pipeline@2026.3 answers the accountability question durably in a way that a named individual does not.

Keep identifiers resolvable, or admit that they are opaque. A URI in a namespace you control, which returns something when dereferenced, is genuinely useful to a downstream consumer. A URI in a namespace you invented that resolves to nothing is a string with punctuation, and it is better to use an explicitly opaque identifier scheme than to imply a resolvability that does not exist.

Get these four decisions right and the graph answers questions for years. Get the first one wrong and no amount of subsequent care recovers it, because the derivation edges were never meaningful to begin with.

# Related

• Spatial Data Lineage & Provenance Tracking — the parent guide with the full model, ISO 19115 projection, and CI patterns
• Spatial Data Audit Reporting & Compliance Governance — the governance program this lineage feeds
• Audit Trail & Evidence Retention — hash the serialized provenance to make it tamper-evident
• DCAT-AP Spatial Profile Mapping — consumes the same PROV vocabulary in published catalog records