GDPR privacy by design for engineers: Articles 25, 32 and 35 in code

{
  "app.accounts.email": {
    "category": "contact",
    "purposes": [
      "account",
      "security-notices"
    ],
    "basis": "Art. 6(1)(b) contract",
    "retention": "30 days after account closure"
  },
  "app.accounts.created_at": {
    "category": "account metadata",
    "purposes": [
      "account"
    ],
    "basis": "Art. 6(1)(b) contract",
    "retention": "30 days after account closure"
  },
  "app.checkins.symptoms": {
    "category": "health",
    "purposes": [
      "symptom-tracking"
    ],
    "basis": "Art. 9(2)(a) explicit consent",
    "retention": "2 years after the check-in",
    "dpia": "DPIA-2027-04"
  },
  "app.checkins.user_id": {
    "category": "identifier",
    "purposes": [
      "symptom-tracking"
    ],
    "basis": "Art. 6(1)(b) contract",
    "retention": "2 years after the check-in"
  }
}

"""Fail the build when a database column holds data the personal-data inventory doesn't describe.

Usage: psql -At -c "SELECT table_schema || '.' || table_name || '.' || column_name
                    FROM information_schema.columns WHERE table_schema = 'app'" \
         | python3 check_inventory.py personal-data.json
Columns that hold no personal data are listed with "category": "none".
"""

import json
import sys

REQUIRED = ("category", "purposes", "basis", "retention")
# GDPR Art. 9(1) special categories
SPECIAL = {
    "racial or ethnic origin",
    "political opinions",
    "religious or philosophical beliefs",
    "trade union membership",
    "genetic",
    "biometric",
    "health",
    "sex life or sexual orientation",
}

with open(sys.argv[1], encoding="utf-8") as f:
    inventory = json.load(f)
columns = {line.strip() for line in sys.stdin if line.strip()}

problems = []
for column in sorted(columns - inventory.keys()):
    problems.append(f"{column}: not in the inventory; describe it or mark it category none")
for column, entry in sorted(inventory.items()):
    if column not in columns:
        problems.append(f"{column}: in the inventory but not in the database; remove the entry")
    if entry.get("category") == "none":
        continue
    for field in REQUIRED:
        if not entry.get(field):
            problems.append(f"{column}: no {field}")
    if entry.get("category") in SPECIAL:
        if not str(entry.get("basis", "")).startswith("Art. 9(2)"):
            problems.append(f"{column}: special category data needs an Art. 9(2) condition")
        if not entry.get("dpia"):
            problems.append(f"{column}: special category data needs a DPIA reference")

print("\n".join(problems) or f"inventory OK: {len(columns)} columns")
sys.exit(1 if problems else 0)

"""Screen a feature for a GDPR data protection impact assessment before design sign-off.

The criteria are the nine in the EDPB-endorsed WP248 rev.01 guidelines; Article 35(3) lists
cases that always need one. Your supervisory authority's Article 35(4) list can add more.
Usage: python3 dpia_screen.py feature.json
"""

import json
import sys

ART_35_3 = {
    "automated-evaluation-with-legal-effects": "Art. 35(3)(a)",
    "large-scale-special-category-or-criminal-data": "Art. 35(3)(b)",
    "large-scale-monitoring-of-public-areas": "Art. 35(3)(c)",
}
WP248 = {
    "evaluation-or-scoring",
    "automated-decision-with-legal-or-similar-effect",
    "systematic-monitoring",
    "sensitive-or-highly-personal-data",
    "large-scale",
    "matching-or-combining-datasets",
    "vulnerable-data-subjects",
    "innovative-technology",
    "prevents-exercising-a-right-or-using-a-service",
}


def screen(feature):
    flags = set(feature["flags"])
    unknown = flags - WP248 - ART_35_3.keys()
    if unknown:
        raise ValueError(f"unknown flags: {sorted(unknown)}")
    mandatory = sorted(ART_35_3[f] for f in flags & ART_35_3.keys())
    criteria = sorted(flags & WP248)
    if mandatory or len(criteria) >= 2:
        verdict = "DPIA required before processing starts"
    elif criteria:
        verdict = "privacy review decides; one criterion can be enough"
    else:
        verdict = "no DPIA indicated; check your authority's Art. 35(4) list"
    return {"feature": feature["id"], "verdict": verdict, "art35_3": mandatory, "wp248": criteria}


if __name__ == "__main__":
    with open(sys.argv[1], encoding="utf-8") as f:
        print(json.dumps(screen(json.load(f)), indent=2))

"""Run access and erasure requests across every store that holds personal data."""

import calendar
from datetime import date


def one_month_after(received: date) -> date:
    """GDPR Art. 12(3): answer within one month of receipt.

    The same day next month, or that month's last day when it is shorter.
    """
    year, month = received.year + received.month // 12, received.month % 12 + 1
    last_day = calendar.monthrange(year, month)[1]
    return received.replace(year=year, month=month, day=min(received.day, last_day))


class SubjectRequests:
    def __init__(self, stores, recipients, tombstones):
        # Every system in the personal-data inventory, including exports, caches and indexes.
        self.stores = stores
        # Parties the data was disclosed to, told about erasure under Art. 19.
        self.recipients = recipients
        # Erased subject IDs, kept to re-apply after a backup restore.
        self.tombstones = tombstones

    def access(self, subject_id, received):
        export = {store.name: store.export(subject_id) for store in self.stores}
        return {"due": one_month_after(received).isoformat(), "data": export}

    def erase(self, subject_id, received):
        removed = {store.name: store.erase(subject_id) for store in self.stores}
        self.tombstones.add(subject_id)
        for recipient in self.recipients:
            recipient.notify_erasure(subject_id)
        return {"due": one_month_after(received).isoformat(), "removed": removed}

    def after_restore(self, store):
        """Run before a restored store serves traffic, so erased subjects stay erased."""
        return sum(store.erase(subject_id) for subject_id in sorted(self.tombstones))

import unittest
from datetime import date

from subject_requests import SubjectRequests, one_month_after


class MemoryStore:
    def __init__(self, name, rows):
        self.name, self.rows = name, rows

    def export(self, subject_id):
        return [row for row in self.rows if row["subject"] == subject_id]

    def erase(self, subject_id):
        before = len(self.rows)
        self.rows = [row for row in self.rows if row["subject"] != subject_id]
        return before - len(self.rows)


class Recipient:
    def __init__(self):
        self.notified = []

    def notify_erasure(self, subject_id):
        self.notified.append(subject_id)


class SubjectRequestTest(unittest.TestCase):
    def setUp(self):
        self.db = MemoryStore("postgres", [{"subject": "u1", "email": "a@example.com"}])
        self.export = MemoryStore("analytics-export", [{"subject": "u1", "symptoms": "rash"}])
        self.clinic = Recipient()
        self.requests = SubjectRequests([self.db, self.export], [self.clinic], set())

    def test_deadline_is_one_calendar_month(self):
        self.assertEqual(one_month_after(date(2027, 3, 31)), date(2027, 4, 30))
        self.assertEqual(one_month_after(date(2027, 1, 31)), date(2027, 2, 28))
        self.assertEqual(one_month_after(date(2027, 12, 15)), date(2028, 1, 15))

    def test_access_covers_every_store(self):
        result = self.requests.access("u1", date(2027, 3, 31))
        self.assertEqual(set(result["data"]), {"postgres", "analytics-export"})
        self.assertEqual(result["due"], "2027-04-30")

    def test_erasure_reaches_every_store_and_recipient(self):
        result = self.requests.erase("u1", date(2027, 3, 31))
        self.assertEqual(result["removed"], {"postgres": 1, "analytics-export": 1})
        self.assertEqual(self.clinic.notified, ["u1"])

    def test_restored_backup_does_not_resurrect_erased_subjects(self):
        self.requests.erase("u1", date(2027, 3, 31))
        restored = MemoryStore("postgres", [{"subject": "u1", "email": "a@example.com"}])
        self.assertEqual(self.requests.after_restore(restored), 1)
        self.assertEqual(restored.export("u1"), [])


if __name__ == "__main__":
    unittest.main()