Mapping GDPR Controls to Lineage Fields

Part of: GDPR for Geospatial Data

When a supervisory authority or your own data protection officer asks for the Article 30 record of processing activities, you should be able to generate it from the lineage store rather than maintain it by hand in a spreadsheet that drifts out of date within weeks. This how-to produces a concrete crosswalk from seven core GDPR articles to specific fields on a geospatial lineage record, then implements a typed model that emits a compliant record of processing directly from lineage rows. It is the field-level companion to the GDPR for geospatial data guide and applies the control-to-field philosophy set out in the regulatory compliance and standards mapping overview.

Prerequisites

• Python 3.10+ with pydantic 2.6+ installed
• A lineage table exposing the columns named in the crosswalk below (subject_key, lawful_basis, purpose, data_categories, recipients, transfer_destination, retention_until, valid_from, content_hash)
• Read access to that store (for example the PostGIS schema from the structuring JSON/XML lineage documents guide)
• A processing register naming each purpose your pipeline exercises

The crosswalk

Each article decomposes into the smallest queryable piece of evidence that demonstrates it. That piece is a lineage field, not a policy paragraph.

The value of expressing it this way is that every cell on the right is something a query can return. If an article decomposes to a field your schema lacks, you have found a gap before an assessor does.

Two articles deserve particular attention for geospatial work. Article 30 is the one most teams underestimate: it is not a one-time document but a living register that must reflect the processing actually happening, which is precisely why deriving it from lineage rows beats maintaining it by hand — the register cannot drift from reality if it is generated from reality. Article 44 is the one location data trips over most often, because a coordinate dataset frequently transits a cloud region or a third-party geocoding service in another jurisdiction without anyone recording the crossing. Binding transfer_destination and transfer_safeguard to the export activity turns that invisible crossing into a logged event, so a transfer without a lawful safeguard becomes detectable rather than latent.

The third column is the one that determines whether generation is possible at all. Controller identity, recipient categories and transfer safeguards are organisational facts the pipeline has no way to observe, and teams frequently stall here — concluding that because some elements cannot be derived, the whole record must stay manual. The resolution is a small versioned policy register keyed by system or dataset class, holding exactly those fields, that the generator joins against. It changes rarely, it is reviewed by the people who own those facts, and it removes the last argument for maintaining the record by hand.

The middle column has a subtlety worth naming: data categories and subject categories are properties of the dataset, not of any single activity, which means they must be assigned at ingestion rather than inferred later. A dataset whose category was never recorded will generate an Article 30 entry with a blank where the regulation requires content, and back-filling it months later is guesswork dressed as documentation.

Implementation

The model below reads lineage rows and emits an Article 30 record of processing. It uses a pydantic model per activity and an aggregator that groups activities into the register entries Article 30 expects — one entry per distinct purpose, listing the categories, recipients, transfers, and retention actually observed.

from __future__ import annotations
from collections import defaultdict
from datetime import datetime
from pydantic import BaseModel, Field


class LineageRow(BaseModel):
    """One append-only lineage event as read from the store."""
    subject_key: str
    activity: str
    lawful_basis: str                       # GDPR Art. 6
    purpose: str                            # GDPR Art. 5(1)(b)
    data_categories: list[str]              # e.g. ["home_geocode", "gps_trace"]
    recipients: list[str] = Field(default_factory=list)
    transfer_destination: str | None = None  # ISO country code, or None
    transfer_safeguard: str | None = None     # e.g. "SCC", "adequacy"
    retention_until: datetime
    valid_from: datetime
    content_hash: str


class Article30Entry(BaseModel):
    """One record-of-processing entry, keyed on purpose (Art. 30(1))."""
    purpose: str
    lawful_bases: set[str]
    data_categories: set[str]
    recipients: set[str]
    transfer_destinations: set[str]
    transfer_safeguards: set[str]
    max_retention: datetime
    activity_count: int
    subject_count: int


def build_article_30(rows: list[LineageRow]) -> list[Article30Entry]:
    """Aggregate lineage rows into an Article 30 record of processing."""
    buckets: dict[str, list[LineageRow]] = defaultdict(list)
    for row in rows:
        buckets[row.purpose].append(row)

    entries: list[Article30Entry] = []
    for purpose, group in buckets.items():
        destinations = {r.transfer_destination for r in group
                        if r.transfer_destination}
        safeguards = {r.transfer_safeguard for r in group
                      if r.transfer_safeguard}
        # Art. 44: any cross-border transfer must carry a safeguard.
        if destinations and not safeguards:
            raise ValueError(
                f"purpose '{purpose}' transfers to {destinations} "
                f"without an Art. 44 safeguard")
        entries.append(Article30Entry(
            purpose=purpose,
            lawful_bases={r.lawful_basis for r in group},
            data_categories={c for r in group for c in r.data_categories},
            recipients={rec for r in group for rec in r.recipients},
            transfer_destinations=destinations,
            transfer_safeguards=safeguards,
            max_retention=max(r.retention_until for r in group),
            activity_count=len(group),
            subject_count=len({r.subject_key for r in group}),
        ))
    return entries

The raise inside the aggregator is deliberate: it turns an Article 44 gap — a cross-border transfer with no recorded safeguard — into a hard failure at report-generation time rather than a finding during an audit. The aggregation shape mirrors Article 30(1)'s own structure, which asks for processing to be described per purpose rather than per record, so grouping on purpose is not an arbitrary key choice but a direct reflection of the regulation’s reporting unit. The set fields collapse the many activities that share a purpose into the distinct categories, recipients, and destinations that a register entry must enumerate, while activity_count and subject_count supply the scale indicators a supervisory authority uses to gauge risk. Because the model is a plain pydantic structure, the same objects serialize straight to the JSON an audit evidence package expects, with no intermediate transcription step where errors creep in.

Be ready for that first comparison, because it is where the effort most often gets abandoned. A carefully maintained spreadsheet shows every cell filled, and the initial generated record shows blanks wherever a lineage field was never populated — which reads, to a stakeholder seeing both, as a step backwards. The honest framing is that the spreadsheet’s completeness was an assertion by whoever last edited it, while the generated record’s gaps are measurements. Publishing the gap count each run turns the argument into a burn-down.

Verification

Confirm the model produces a defensible record by feeding it two rows for the same purpose, one with an unsafeguarded transfer, and asserting behavior.

from datetime import datetime, timezone

base = dict(subject_key="s-001", activity="geocode",
            lawful_basis="public_task", purpose="address_validation",
            data_categories=["home_geocode"],
            retention_until=datetime(2027, 1, 1, tzinfo=timezone.utc),
            valid_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
            content_hash="ab12")

ok = LineageRow(**base)
entries = build_article_30([ok])
assert entries[0].subject_count == 1
assert entries[0].max_retention.year == 2027
print("Art. 30 entry for:", entries[0].purpose)

A passing run prints the purpose and asserts the aggregation counted one subject and carried the retention ceiling through. To confirm the Article 44 guard, add a row with transfer_destination="US" and transfer_safeguard=None and check that build_article_30 raises. The generated Article30Entry list is your record of processing; serialize it to JSON for the DPO or into the audit evidence package.

Keeping the Crosswalk Current

Constraining purpose to an enumeration is the single change that prevents the most common form of drift. Free-text purposes accumulate variants — “flood modelling”, “flood modeling”, “hydrology analysis” — that all mean the same thing to a human and are three distinct values to a generator, producing an Article 30 record listing three processing activities where one exists. An enum forces the conversation about whether a genuinely new purpose has appeared at the moment it appears, which is when someone still knows the answer.

The third row cannot be automated and should not be pretended otherwise. Schedule a review of the crosswalk against current supervisory guidance on the same cadence as other policy reviews, version the file, and record the date and reviewer. That turns “the crosswalk is current” from an assumption into a dated statement someone is accountable for.

Gotchas & edge cases

• Purpose sprawl inflates the register. Free-text purpose values that differ only by spelling (“address validation” vs “address_validation”) produce duplicate Article 30 entries. Normalize purpose against the processing register before aggregation, ideally with an enum, so one real purpose maps to one entry.
• subject_key must be an opaque token, not raw PII. If the key is an email or a device ID, the Article 30 record you generate becomes a new copy of personal data. Keep the token-to-identity map in a separate vault, as covered in the parent GDPR for geospatial data guide.
• Erased subjects still appear in aggregates. After a crypto-shred, subject_key rows persist for audit but carry no payload. Count them for volume metrics, but never attempt to resolve their data_categories back to a person — that data is gone by design.

Related

• GDPR for Geospatial Data — the regime-level treatment
• Anonymizing Location Data for GDPR — recording a discharge, not just a transformation
• Compliance Framework Mapping — the general control-to-field method
• FISMA Control Mapping for GIS Pipelines — the same technique for a different regime
• Part of: GDPR for Geospatial Data