Geospatial Compliance Metadata Automation Hub

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.