How to Map DSR Types to GDPR Article 15
A Data Subject Request (DSR) that invokes the right of access almost never arrives labelled “Article 15.” It arrives as “export my profile,” “what data do you hold on me?” or “send me a copy of my account history” — free text that must be translated into a single canonical intent before any data aggregation runs. This page is for the privacy engineer building the classification step that sits inside the GDPR vs CCPA Request Taxonomies vocabulary layer and feeds the wider DSR Architecture & Intake Routing pipeline. Get the mapping wrong and the failure is expensive: a misclassified access request either starts the wrong statutory clock, over-collects data the requester never asked for, or silently routes to a workflow with the wrong exemptions applied — each a reportable defect under GDPR Art. 12(3), which obliges the controller to act only on a verifiable request.
The precise gap this page closes is the distance between an ambiguous natural-language submission and a signed, jurisdiction-tagged ARTICLE_15_ACCESS record that a downstream fulfillment worker can execute without re-parsing untrusted input. The mapping must be deterministic (the same payload always yields the same intent), tamper-evident (a forged webhook body is rejected before classification), and fully auditable (every classification decision leaves a correlation ID in an append-only log).
Prerequisites
- Python 3.11+ — for
zoneinfoin the standard library and modern typing syntax used in the SLA computation. - Pydantic v2 (
pip install "pydantic>=2.6") — strict schema validation andfield_validatorfor payload normalization at the ingestion boundary. See the Pydantic validation documentation for strict-mode semantics. - tenacity (
pip install tenacity) — bounded, idempotent retries around transient downstream faults. - Standard library
hmac,hashlib,re, andlogging— no third-party crypto is required for HMAC-SHA256 request authentication. - A shared HMAC secret provisioned to both the intake origin (public form or partner API) and this service, stored in a secrets manager and never in source.
- A configured set of GDPR regions — the two-letter region codes for which Article 15 applies, so a non-EEA submission can fall back rather than be misrouted.
This mapping runs after the Secure Intake Form Design perimeter has produced a raw payload, and before the Jurisdiction Routing Logic dispatches the classified record to a jurisdiction-specific queue.
Step-by-step implementation
The routing gate rejects, falls back, or approves through a deterministic sequence of checks:
Step 1: Define the canonical intent enum
Users rarely cite legal articles verbatim, so the first job is to collapse every phrasing onto a small, closed set of canonical intents. A string-backed Enum makes the intent both type-safe in code and serializable into the audit log.
from enum import Enum
class DSRIntent(str, Enum):
"""Closed vocabulary of DSR intents this pipeline recognizes."""
ARTICLE_15_ACCESS = "article_15_access" # GDPR Art. 15 — right of access
ARTICLE_17_ERASURE = "article_17_erasure" # GDPR Art. 17 — right to erasure
ARTICLE_20_PORTABILITY = "article_20_portability" # GDPR Art. 20 — portability
UNKNOWN = "unknown" # requires manual classification
Only ARTICLE_15_ACCESS proceeds to aggregation on this path; the other members exist so that a request which is not an access request is detected and re-routed rather than executed as one. UNKNOWN deliberately pauses the SLA clock until a human resolves it — never guess a statutory basis.
Step 2: Normalize the payload at the schema boundary
Normalization must be deterministic and happen inside validation, so no downstream code ever sees raw markup or inconsistent casing. Pydantic v2’s field_validator strips HTML and lowercases the free-text payload as the model is constructed.
import re
from typing import Optional
from pydantic import BaseModel, ConfigDict, field_validator
class DSRIntake(BaseModel):
"""Validated, normalized intake record produced at the ingestion boundary."""
model_config = ConfigDict(strict=True, frozen=True, extra="forbid")
request_id: str
jurisdiction: str
raw_payload: str
hmac_signature: Optional[str] = None
@field_validator("raw_payload")
@classmethod
def normalize_payload(cls, value: str) -> str:
"""Strip HTML tags, lowercase, and trim so classification is deterministic."""
return re.sub(r"<[^>]+>", "", value).lower().strip()
strict=True prevents silent type coercion (a jurisdiction of 45 will not slip through as a string), frozen=True guarantees no later stage mutates the classified record, and extra="forbid" rejects unexpected fields that could smuggle in unvalidated data.
Step 3: Classify intent deterministically
Classification is regex-driven against the normalized text. Ordering the patterns matters: the first match wins, so the most specific access vocabulary is evaluated before broader terms. Anything that matches nothing falls to UNKNOWN rather than defaulting to access.
import hashlib
import hmac
import logging
logger = logging.getLogger("dsr.pipeline")
class Article15Router:
"""Maps normalized DSR payloads onto the canonical intent vocabulary."""
INTENT_PATTERNS: dict[DSRIntent, re.Pattern[str]] = {
DSRIntent.ARTICLE_15_ACCESS: re.compile(
r"(access|download|export|copy|show.*data|what.*(know|hold)|account.*(details|history))"
),
DSRIntent.ARTICLE_17_ERASURE: re.compile(r"(delete|erase|remove|forget|wipe)"),
DSRIntent.ARTICLE_20_PORTABILITY: re.compile(
r"(port|transfer|machine.?readable|csv|json.*export|move.*data)"
),
}
def __init__(self, config: dict) -> None:
self.allowed_jurisdictions: set[str] = set(config.get("gdpr_regions", []))
self.hmac_secret: bytes = config.get("hmac_secret", "").encode()
def classify_intent(self, payload: str) -> DSRIntent:
"""Return the first canonical intent whose pattern matches, else UNKNOWN."""
for intent, pattern in self.INTENT_PATTERNS.items():
if pattern.search(payload):
return intent
return DSRIntent.UNKNOWN
Step 4: Verify authenticity and derive a correlation ID
Before trusting the payload, verify the HMAC-SHA256 signature over the raw body with a constant-time comparison (hmac.compare_digest) to defeat both forgery and timing side-channels — the same perimeter control the Secure Intake Form Design stage enforces. The correlation ID is a stable hash of the request ID and payload; it doubles as an idempotency key so a duplicate submission returns the existing record instead of resetting the SLA clock.
class Article15Router: # continued from Step 3
def _verify_hmac(self, payload: str, signature: Optional[str]) -> bool:
"""Constant-time HMAC-SHA256 verification of the raw payload."""
if not signature or not self.hmac_secret:
return False
expected = hmac.new(
self.hmac_secret, payload.encode(), hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)
def correlation_id(self, request_id: str, payload: str) -> str:
"""Deterministic idempotency key spanning the request id and payload."""
seed = f"{request_id}:{payload}".encode()
return hashlib.sha256(seed).hexdigest()[:16]
Step 5: Gate, route, and anchor the SLA
The gate runs the checks in order — authenticity, then jurisdiction, then intent — and only an authenticated, in-scope ARTICLE_15_ACCESS request is approved. GDPR Art. 12(3) sets a one-month response window (extendable by two further months for complex requests), which this code anchors as a UTC deadline the moment classification succeeds. Transient connection faults are retried with bounded exponential backoff via tenacity; a genuine auth failure raises immediately and is never retried.
from datetime import datetime, timedelta, timezone
from tenacity import (
retry,
retry_if_exception_type,
stop_after_attempt,
wait_exponential,
)
class SignatureError(ValueError):
"""Raised when a submission fails HMAC verification."""
class Article15Router: # continued from Step 4
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type(ConnectionError),
reraise=True,
)
def route_request(self, intake: DSRIntake) -> dict:
"""Classify and gate a normalized intake record onto the Article 15 path."""
cid = self.correlation_id(intake.request_id, intake.raw_payload)
if not self._verify_hmac(intake.raw_payload, intake.hmac_signature):
logger.warning("auth failure cid=%s", cid)
raise SignatureError("Invalid cryptographic signature")
if intake.jurisdiction not in self.allowed_jurisdictions:
logger.info("jurisdiction fallback cid=%s region=%s", cid, intake.jurisdiction)
return {"status": "routed_to_fallback", "correlation_id": cid}
intent = self.classify_intent(intake.raw_payload)
if intent is not DSRIntent.ARTICLE_15_ACCESS:
logger.info("non-article-15 intent cid=%s intent=%s", cid, intent.value)
return {"status": "rerouted", "intent": intent.value, "correlation_id": cid}
deadline = datetime.now(timezone.utc) + timedelta(days=30)
logger.info("article 15 approved cid=%s deadline=%s", cid, deadline.isoformat())
return {
"status": "approved",
"intent": intent.value,
"correlation_id": cid,
"sla_deadline": deadline.isoformat(),
}
Configuration reference
Common phrasings map onto the closed intent vocabulary as follows:
| User phrasing (examples) | Canonical intent | GDPR basis | Response window |
|---|---|---|---|
| “access”, “download”, “export”, “copy”, “what data do you hold” | ARTICLE_15_ACCESS |
Art. 15 — right of access | 1 month (Art. 12(3)) |
| “delete”, “erase”, “remove”, “forget” | ARTICLE_17_ERASURE |
Art. 17 — right to erasure | 1 month (Art. 12(3)) |
| “port”, “transfer”, “machine-readable”, “move my data” | ARTICLE_20_PORTABILITY |
Art. 20 — portability | 1 month (Art. 12(3)) |
| unrecognized | UNKNOWN |
manual classification | clock paused |
The router itself is driven by these parameters:
| Parameter | Type | Default | Compliance note |
|---|---|---|---|
gdpr_regions |
set[str] |
set() |
Region codes for which Art. 15 applies; a code outside this set falls back rather than misrouting a non-EEA subject. |
hmac_secret |
bytes |
b"" |
Shared secret for HMAC-SHA256; an empty secret fails verification closed per GDPR Art. 12(3). |
| SLA base window | timedelta |
30 days |
GDPR Art. 12(3) “one month”; extendable by up to two further months for complex requests. |
retry stop_after_attempt |
int |
3 |
Bounds transient retries so a stuck request cannot silently consume the SLA window. |
Verification
Confirm the mapping is deterministic with a small payload matrix. Each phrasing must resolve to exactly one canonical intent, and an unrecognized string must resolve to UNKNOWN — never to access.
import pytest
router = Article15Router({"gdpr_regions": {"DE", "FR", "IE"}, "hmac_secret": "s3cret"})
@pytest.mark.parametrize(
"payload,expected",
[
("please export a copy of my profile", DSRIntent.ARTICLE_15_ACCESS),
("what data do you hold on me?", DSRIntent.ARTICLE_15_ACCESS),
("delete my account and forget me", DSRIntent.ARTICLE_17_ERASURE),
("transfer my data to another provider", DSRIntent.ARTICLE_20_PORTABILITY),
("i have a billing question", DSRIntent.UNKNOWN),
],
)
def test_classify_intent(payload: str, expected: DSRIntent) -> None:
assert router.classify_intent(payload) == expected
You should observe a single structured log line per decision — article 15 approved cid=... deadline=... for an approved access request — and the approved result must carry both a correlation_id and an ISO-8601 sla_deadline. Re-submitting the same request_id and payload must yield the identical correlation_id, proving idempotency and confirming the SLA clock is not reset. Every classification decision, keyed by correlation ID, is written to an append-only audit log so the mapping satisfies examination under GDPR Art. 30.
Troubleshooting
Access requests silently classified as UNKNOWN.
Root cause: the intake vocabulary uses phrasing the regex does not cover (for example “give me my records”). Fix: extend the ARTICLE_15_ACCESS pattern and add the phrase to the test matrix so the regression is captured; never widen a pattern without a test.
A portability request matched as access.
Root cause: pattern ordering — “export” appears in both the access and portability vocabularies, and access is evaluated first. Fix: qualify the access pattern (require “copy”/“what data”) or move the more specific machine.?readable/transfer portability check ahead of the generic export term.
Every request rejected with auth failure.
Root cause: the hmac_secret is empty, or the signature was computed over the normalized payload while the router verifies against the raw body (or vice-versa). Fix: sign and verify the exact same bytes, and confirm the secret is loaded from the secrets manager rather than defaulting to b"", which fails closed.
Legitimate EEA request routed to fallback.
Root cause: the subject’s region code is not in gdpr_regions (for example an EEA state added after deployment). Fix: treat the region set as configuration, validate held-out region codes against the matrix before enabling them, and fail closed to manual review rather than approving an unmapped region.
Duplicate submissions resetting the SLA clock.
Root cause: the correlation ID is derived from a value that varies per attempt (a timestamp or a fresh UUID). Fix: derive it only from the stable request_id and normalized payload, as in correlation_id, so retries and duplicates collapse onto one record.
Related
- GDPR vs CCPA Request Taxonomies — the parent vocabulary this mapping classifies against
- Handling CCPA Deletion vs Opt-Out Requests — the sibling mapping problem on the California side of the taxonomy
- Secure Intake Form Design — the HMAC-signed, validated record this classifier consumes
- Building a Jurisdiction-Aware Intake Router in Python — how the classified record is dispatched to a jurisdiction-specific queue
- DSR Architecture & Intake Routing — the end-to-end pipeline this classification step sits inside