Validating JSON Payloads Against DSR Schemas

Data Subject Request (DSR) pipelines run under statutory clocks — the “without undue delay and in any event within one month” mandate of GDPR Article 12(3) and the 45-day baseline of CCPA §1798.130(a)(2) — and a single malformed payload entering ingestion can stall automated fulfillment, breach an SLA, and leave an accountability gap under GDPR Article 5(2). This page shows how to build the concrete validator that enforces the contract defined in Schema Validation Rules, the deterministic gate that sits between raw connector output and the staged manifest inside Cross-System Data Discovery & Sync. The engineer who reaches this page has a schema and a stream of imperfect real-world payloads, and needs a validator that is a pure function: identical inputs always yield identical accept/reject decisions, so every verdict can be replayed during an audit and no raw PII ever leaks into a log line.

Validation here is not a passive syntax check — it is a compliance control point. It must enforce strict type boundaries, reconcile jurisdictional consent semantics, and guarantee that request scopes honor data minimization under GDPR Article 5(1)© before any downstream processing begins. The flow below moves a payload from an untrusted external shape to a validated, structured verdict that is either published to fulfillment or routed — with masked identifiers — to a dead-letter queue.

Prerequisites #

• Python 3.11+ — the code uses PEP 604 unions and modern type hints.
• Pydantic v2 (pydantic>=2.6) for high-performance runtime validation using ConfigDict, field_validator, and model_validator.
• jsonschema (jsonschema>=4.21) for the declarative draft 2020-12 contract that gives auditors a language-independent artifact.
• Standard library only for the security primitives: hashlib for SHA-256 tokenization of subject identifiers and logging for structured, PII-free diagnostics.
• Infrastructure: a message broker with a dead-letter queue (dsr.dlq.manual) and a compliance-review queue, plus an append-only (WORM) store for the accept/reject ledger. Connection details arrive through environment variables, never hardcoded.

This page assumes the ingestion contract and native-type mappings from Database Connector Configuration already exist; the validator below enforces that contract at the boundary rather than defining it.

Step-by-step implementation #

Step 1 — Establish the base contract and safe PII masking #

Start with an immutable baseline model. strict=True disables Pydantic’s implicit coercion so type violations surface as explicit errors rather than silent conversions, and extra="forbid" enforces data minimization at the boundary per GDPR Article 5(1)© — any undeclared field is rejected, never persisted into a regulated pipeline. Every diagnostic that touches an identifier is masked first: never log a raw subject_id.

import hashlib
import logging

from pydantic import BaseModel, ConfigDict
from pydantic_core import PydanticCustomError

logger = logging.getLogger("dsr_validator")


def mask_pii(value: str) -> str:
    """Return a stable, non-reversible token so logs never carry raw identifiers."""
    digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:8]
    return f"[REDACTED:{digest}]"


class DSRPayloadBase(BaseModel):
    """Immutable baseline contract shared by every DSR request variant."""

    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)

    request_id: str
    subject_id: int
    request_type: str

Step 2 — Coerce ambiguous types deterministically #

Production payloads rarely match documentation: numeric identifiers frequently arrive as strings ("subject_id": "10482"), which a strict validator rejects outright. Rather than relaxing strictness globally, add a narrow field_validator(mode="before") that attempts a safe coercion, logs the masked transformation for the audit trail, and rejects anything ambiguous ("10482abc"). Coercion must never alter semantic meaning.

from pydantic import field_validator


class DSRPayloadCoerced(DSRPayloadBase):
    """Base contract with audited, safe numeric coercion for subject_id."""

    @field_validator("subject_id", mode="before")
    @classmethod
    def coerce_numeric_id(cls, v: object) -> object:
        """Coerce a clean numeric string to int; reject non-numeric strings."""
        if isinstance(v, str):
            logger.info("Coercing subject_id from string: %s", mask_pii(v))
            try:
                return int(v)
            except ValueError as exc:
                raise PydanticCustomError(
                    "invalid_numeric_string",
                    "subject_id must resolve to a valid integer",
                ) from exc
        return v

Step 3 — Enforce scope, uniqueness, and non-empty categories #

data_categories and processing_purposes routinely arrive with mixed types, duplicates, or empty objects that violate data minimization. Constrain the field to an enumerated allowlist with a set type — which deduplicates automatically — then reject an empty scope, because a request that names no data category cannot be lawfully fulfilled under the purpose-limitation principle of GDPR Article 5(1)(b).

from typing import Literal


DataCategory = Literal["contact_info", "financial_records", "location_data", "biometric_data"]


class DSRPayloadWithCategories(DSRPayloadCoerced):
    """Adds a deduplicated, allowlisted, non-empty data-category scope."""

    data_categories: set[DataCategory]

    @field_validator("data_categories")
    @classmethod
    def enforce_non_empty(cls, v: set[str]) -> set[str]:
        """Reject a request whose scope names no data category."""
        if not v:
            raise PydanticCustomError(
                "empty_scope",
                "data_categories must contain at least one valid scope",
            )
        return v

Step 4 — Evaluate conditional and null-versus-missing rules #

Two of the hardest DSR validation cases are cross-field dependencies and the difference between an explicit null and an omitted field. legal_basis is mandatory only when request_type == "erasure" (the record required by GDPR Article 17), and a consent_timestamp that is acceptable to omit under CCPA can be fatal under GDPR. Resolve both with model_validator(mode="after"), which runs after baseline parsing and gives typed, field-level errors — mirroring JSON Schema conditional validation while keeping runtime granularity.

from pydantic import model_validator


class DSRPayload(DSRPayloadWithCategories):
    """Full request contract with conditional legal-basis and consent rules."""

    request_type: Literal["access", "erasure", "rectification"]
    jurisdiction: Literal["GDPR", "CCPA", "LGPD"]
    legal_basis: str | None = None
    consent_timestamp: str | None = None

    @model_validator(mode="after")
    def enforce_cross_field_rules(self) -> "DSRPayload":
        """Apply erasure-basis and GDPR-consent rules after baseline parsing."""
        if self.request_type == "erasure" and not self.legal_basis:
            raise PydanticCustomError(
                "missing_legal_basis",
                "legal_basis is mandatory for erasure requests (GDPR Art. 17)",
            )
        if self.jurisdiction == "GDPR" and self.consent_timestamp is None:
            raise PydanticCustomError(
                "gdpr_consent_missing",
                "GDPR payloads require an explicit consent_timestamp (cannot be null or omitted)",
            )
        return self

Step 5 — Produce a masked verdict and route deterministically #

Deterministic validation is only half the pipeline: when a payload fails, it must be routed without exposing PII, without triggering infinite retries, and without corrupting downstream state. Parse the ValidationError into a structured envelope containing field_path, error_code, severity, and a hashed subject token, then route by severity so invalid payloads never block valid requests. Immediate structural violations and version mismatches go to the dead-letter queue; softer failures go to human compliance review.

from typing import Any

from pydantic import ValidationError


def validate_payload(raw: dict[str, Any]) -> dict[str, Any]:
    """Return a structured, PII-masked verdict for one raw DSR payload."""
    try:
        payload = DSRPayload(**raw)
        return {"status": "valid", "request_id": payload.request_id}
    except ValidationError as exc:
        first = exc.errors()[0]
        token = mask_pii(str(raw.get("subject_id", "")))
        return {
            "status": "invalid",
            "field_path": ".".join(str(p) for p in first["loc"]),
            "error_code": first["type"],
            "severity": "HIGH" if first["type"] in {"extra_forbidden", "missing"} else "LOW",
            "subject_token": token,
        }


def route_validation_result(result: dict[str, Any]) -> str:
    """Map a verdict to a queue; failing closed for high-severity errors."""
    if result["status"] == "valid":
        return "publish_to_fulfillment_queue"
    error_code = result.get("error_code", "UNKNOWN")
    severity = result.get("severity", "LOW")
    if severity in {"HIGH", "CRITICAL"} or "schema_version_mismatch" in error_code:
        return "route_to_dlq"
    if "transient_dependency" in error_code:
        return "retry_with_backoff"
    return "route_to_compliance_review"

Configuration reference #

Verification #

Confirm the validator behaves as a pure function before trusting it with real subject data. Assert that a clean numeric string coerces, that a malformed one is rejected with the expected code, that a GDPR payload without a consent timestamp fails, and that no verdict ever carries a raw identifier.

def test_safe_numeric_coercion() -> None:
    """A clean numeric string must coerce; verdict must be valid."""
    result = validate_payload(
        {
            "request_id": "r-1",
            "subject_id": "10482",
            "request_type": "access",
            "jurisdiction": "CCPA",
            "data_categories": ["contact_info"],
        }
    )
    assert result["status"] == "valid"


def test_gdpr_requires_consent_timestamp() -> None:
    """A GDPR payload with no consent_timestamp must fail with a typed code."""
    result = validate_payload(
        {
            "request_id": "r-2",
            "subject_id": 10482,
            "request_type": "access",
            "jurisdiction": "GDPR",
            "data_categories": ["contact_info"],
        }
    )
    assert result["status"] == "invalid"
    assert route_validation_result(result) in {"route_to_dlq", "route_to_compliance_review"}


def test_verdict_never_leaks_raw_identifier() -> None:
    """An invalid verdict must carry only a masked token, never the raw id."""
    result = validate_payload(
        {"request_id": "r-3", "subject_id": 999, "request_type": "access"}
    )
    assert result["status"] == "invalid"
    assert result["subject_token"].startswith("[REDACTED:")
    assert "999" not in result["subject_token"]

Expected log output for a coerced identifier is a single masked line such as Coercing subject_id from string: [REDACTED:9f86d081] — never the raw value. The compliance assertion to hold under load: every payload produces exactly one verdict, every invalid verdict maps to exactly one queue via route_validation_result, and no valid request is ever blocked by an invalid neighbor.

Troubleshooting #

Every payload with a nested object is rejected Root cause: extra="forbid" (or additionalProperties: false) inherited on a metadata sub-object that must accept arbitrary keys. Fix: relax the constraint on that specific object or enumerate its allowed properties explicitly; do not silently inherit forbid on free-form metadata.

Clean numeric strings fail validation Root cause: strict=True disables coercion globally, so "10482" never becomes 10482. Fix: add the narrow field_validator(mode="before") from Step 2 rather than turning strict mode off, so only audited, safe coercions are allowed.

Conditional rules never fire Root cause: cross-field logic written as a field_validator, which cannot see sibling fields. Fix: move dependency checks into a model_validator(mode="after") so request_type, legal_basis, and consent_timestamp are all populated when the rule runs.

A missing field and an explicit null are treated identically Root cause: relying on a truthiness check that cannot distinguish absence from None. Fix: type the field as str | None with a default and assert presence explicitly (is None) inside the model validator, since GDPR and CCPA treat the two states differently.

Raw subject identifiers appear in the observability platform Root cause: logging the ValidationError or the raw payload directly. Fix: build the masked envelope from Step 5 first and log only field_path, error_code, severity, and the hashed subject_token.

Related #

• Schema Validation Rules — the parent contract: the deterministic gate this validator enforces at each ingestion point.
• Implementing Celery for Async Polling — where this same Pydantic gate seals every extracted payload before the audit ledger.
• Database Connector Configuration — the native-type-to-JSON-Schema mappings this contract validates against.
• SaaS API Sync Strategies — the transient-versus-permanent failure taxonomy that governs the routing in Step 5.
• Cross-System Data Discovery & Sync — the discovery stage this validation boundary belongs to.