Schema Validation Rules: The Deterministic Gate Before DSR Fulfillment

Within the broader Cross-System Data Discovery & Sync architecture, schema validation is the control point that decides whether a discovered record ever reaches the response generation stage. Discovery fans out across relational stores, warehouses, and SaaS APIs, and every one of those sources returns differently shaped payloads. Without a rigid, deterministic validation boundary at each ingestion point, downstream fulfillment inherits malformed records that trigger silent data loss, incomplete access responses, or audit failures. This is not a passive syntax check — under the GDPR Article 5(2) accountability principle a controller must be able to demonstrate that only structurally sound, in-scope data entered the pipeline, and under GDPR Article 12(3) and CCPA §1798.130(a)(2) the statutory clock keeps running while corrupted payloads stall automated workflows. The validation layer therefore has to reject malformed structures immediately, emit a tamper-evident record of every accept/reject decision, and behave as a pure function so any decision can be replayed during an audit.

This page specifies the validation contract that sits between the raw connector output produced by Database Connector Configuration and SaaS API Sync Strategies, and the deduplicated manifest handed to the PII Extraction & Redaction Pipelines stage. The stages below move a payload from an untrusted external shape to a validated, hashed, audit-stamped record — failing closed to a dead-letter queue on any contract violation.

Phase 1: Ingestion Contracts & Type Coercion #

The first stage establishes an immutable baseline contract. When configuring raw table scans through the database connectors, native SQL types must be mapped deterministically to JSON Schema primitives so that extraction output conforms to a predictable structure before any downstream processing begins. The contract is expressed with JSON Schema (draft 2020-12) for declarative auditability and enforced at runtime with Pydantic v2 for performance and typed error paths.

Note that additionalProperties: false on a nested object such as metadata means no properties are permitted inside it. If the schema must accept arbitrary key-value metadata, either relax that constraint for that object or enumerate the allowed properties explicitly — silently inheriting false is the most common cause of false-positive rejections.

from datetime import datetime, timezone
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator


class DSRIngestionContract(BaseModel):
    """Strict DSR payload contract aligned with GDPR Art. 15 / CCPA 1798.110 scope."""

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

    subject_id: str = Field(pattern=r"^[A-Z0-9]{8,32}$")
    request_type: Literal["access", "deletion", "rectification", "portability"]
    timestamp: datetime
    jurisdiction: str = Field(pattern=r"^[A-Z]{2}$")

    @field_validator("timestamp")
    @classmethod
    def enforce_tz_aware(cls, v: datetime) -> datetime:
        """Reject naive timestamps — SLA math under GDPR Art. 12(3) requires UTC."""
        if v.tzinfo is None:
            raise ValueError("timestamp must be timezone-aware (ISO 8601 with offset)")
        return v.astimezone(timezone.utc)

The equivalent declarative JSON Schema is versioned alongside pipeline code so a supervisory authority can reconstruct exactly which contract was in force on the date a request was processed:

DSR_INGESTION_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "required": ["subject_id", "request_type", "timestamp", "jurisdiction"],
    "properties": {
        "subject_id": {"type": "string", "pattern": "^[A-Z0-9]{8,32}$"},
        "request_type": {"enum": ["access", "deletion", "rectification", "portability"]},
        "timestamp": {"type": "string", "format": "date-time"},
        "jurisdiction": {"type": "string", "pattern": "^[A-Z]{2}$"},
        "metadata": {"type": "object", "additionalProperties": True},
    },
    "additionalProperties": False,
}

Phase 2: Payload Normalization & Transformation #

External SaaS endpoints rarely conform to the internal contract. A lightweight normalization layer standardizes key casing, flattens the wrapper arrays vendors love to nest (data, results, records, payload), and coerces temporal formats before validation fires. Normalization is a deterministic bridge, not a validator — it never accepts or rejects, it only reshapes, so the strict gate in Phase 3 remains the single source of truth for compliance decisions.

from datetime import datetime, timezone
from typing import Any


def normalize_external_payload(raw: dict[str, Any]) -> dict[str, Any]:
    """Reshape a heterogeneous vendor payload into the internal contract shape.

    Pure and side-effect free: identical input always yields identical output,
    which keeps replay testing and audit reconstruction deterministic.
    """
    normalized: dict[str, Any] = {
        k.lower().replace("-", "_").replace(" ", "_"): v for k, v in raw.items()
    }

    for wrapper_key in ("data", "results", "records", "payload"):
        value = normalized.get(wrapper_key)
        if isinstance(value, list):
            normalized[wrapper_key] = value[0] if value else {}
            break

    ts = normalized.get("timestamp")
    if isinstance(ts, (int, float)):
        normalized["timestamp"] = datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()

    return normalized

Phase 3: Strict Validation Engine & Audit Trails #

The core engine rejects malformed structures immediately and emits an immutable audit record for every decision. Compliance officers require exact validation failure paths, not boolean outcomes, so the engine surfaces the JSON path of the first violation and couples every accepted payload with a SHA-256 hash for tamper-evident logging, using an algorithm approved under NIST SP 800-107. The step-by-step debugging patterns for the four payloads that most often reach this gate — type coercion ambiguity, nested-array drift, conditional field dependencies, and null-vs-missing semantics — are worked through in Validating JSON payloads against DSR schemas.

import hashlib
import json
import logging
import uuid
from datetime import datetime, timezone
from typing import Any

from pydantic import ValidationError

logger = logging.getLogger("dsr_validator")


def audit_and_validate(raw_payload: dict[str, Any]) -> DSRIngestionContract:
    """Normalize, validate, and emit a tamper-evident audit record.

    Returns the validated model on success; raises on contract violation after
    logging a REJECTED record. Never logs raw payloads (data minimization).
    """
    trace_id = str(uuid.uuid4())
    normalized = normalize_external_payload(raw_payload)

    try:
        model = DSRIngestionContract.model_validate(normalized)
    except ValidationError as err:
        first = err.errors()[0]
        detail = f"{'.'.join(str(p) for p in first['loc'])}: {first['msg']}"
        logger.critical(
            "AUDIT: %s",
            json.dumps(
                {
                    "trace_id": trace_id,
                    "status": "REJECTED",
                    "error_detail": detail,
                    "rejected_at": datetime.now(timezone.utc).isoformat(),
                }
            ),
        )
        raise

    canonical = model.model_dump_json()
    payload_hash = hashlib.sha256(canonical.encode()).hexdigest()
    logger.info(
        "AUDIT: %s",
        json.dumps(
            {
                "trace_id": trace_id,
                "status": "VALIDATED",
                "ingested_at": datetime.now(timezone.utc).isoformat(),
                "payload_hash": payload_hash,
            }
        ),
    )
    return model

Phase 4: Stateless Validation & Error Routing #

Network instability introduces transient failures that must never contaminate validation metrics. Schema validation stays entirely stateless and decoupled from HTTP status codes; the orchestration layer routes transport errors before any payload reaches the gate. This mirrors the retry-versus-quarantine policy the discovery layer applies in Async Polling & Queue Management: transient faults back off, permanent faults quarantine immediately so they do not burn the statutory window.

from http import HTTPStatus
from typing import Callable

import requests


def route_discovery_response(response: requests.Response, retry_fn: Callable) -> None:
    """Classify a transport response into terminal / transient / rejected states."""
    status = response.status_code

    if status == HTTPStatus.NOT_FOUND:
        # Subject not found at this source: terminal, mark this slice complete.
        logger.warning("Terminal 404 for %s — slice complete", response.url)
        return

    if status >= HTTPStatus.INTERNAL_SERVER_ERROR or status == HTTPStatus.TOO_MANY_REQUESTS:
        # Server-side transient or throttle: exponential backoff.
        logger.info("Transient %s: queuing slice for retry", status)
        retry_fn(response)
        return

    if status == HTTPStatus.BAD_REQUEST:
        # Upstream rejected our request shape: permanent, do not retry.
        logger.error("Upstream schema mismatch: %s", response.text)
        raise ValueError("Upstream schema mismatch detected")

Phase 5: Regulatory Evolution & Schema Versioning #

Privacy regulations change continuously, so validation contracts must adapt without breaking in-flight pipelines. A schema registry validates against the current regulatory draft while retaining fallback validation paths for legacy payloads. Version JSON contracts alongside pipeline code in source control, and constrain changes to backward-compatible additions — new optional fields only — so a jurisdictional update or a newly recognized data subject right never breaks a payload already in the queue. The same taxonomy divergence that scope resolution handles at discovery, described in GDPR vs CCPA Request Taxonomies, surfaces here as conditional validation branches keyed on the jurisdiction field.

from pydantic import BaseModel, ConfigDict


class ContractRegistry:
    """Resolve the validation model for a payload's declared schema version."""

    def __init__(self) -> None:
        self._versions: dict[str, type[BaseModel]] = {}

    def register(self, version: str, model: type[BaseModel]) -> None:
        self._versions[version] = model

    def resolve(self, payload: dict) -> type[BaseModel]:
        # Default to the latest contract when a payload omits its version.
        version = str(payload.get("schema_version", "latest"))
        if version not in self._versions:
            logger.warning("Unknown schema_version %s — falling back to latest", version)
            version = "latest"
        return self._versions[version]

Edge Cases & Conflict Resolution #

The gate is where ambiguous and contradictory inputs must be forced into a deterministic decision rather than a guess.

• Conflicting jurisdiction signals. When the payload’s jurisdiction field disagrees with the residency resolved upstream by the Jurisdiction Routing Logic, the validator does not average them — it trusts the attested routing decision and quarantines the mismatch for review, since a wrong framework selection is a compliance defect, not a data-quality one.
• UNKNOWN jurisdiction fallback. A payload carrying an UNKNOWN jurisdiction is validated against the strictest applicable contract (the GDPR-shaped model), because failing toward the most protective statute is the only defensible default under the GDPR Article 5(1)© minimization principle.
• Null vs missing semantics. GDPR treats an explicit null consent_timestamp differently from an omitted one; the contract distinguishes default=None (present, null) from a required field (must be supplied) so a CCPA-acceptable omission is not silently accepted as GDPR-compliant.
• Multi-jurisdiction overlap. A subject plausibly covered by both frameworks is validated against the union of required fields and timed to the shorter statutory window, consistent with the 30-Day vs 45-Day SLA Mapping.

Performance & Scale Considerations #

Validation runs on every discovered record across every connector, so it sits on the hot path of the whole discovery stage. Compile each JSON Schema and instantiate each Pydantic model class once at worker startup, not per payload — Pydantic v2’s Rust core validates a compiled model an order of magnitude faster than re-parsing a schema on each call. Cache the resolved contract-registry lookup in Redis keyed by schema_version so the registry is not re-walked for high-volume tenants. When discovery is partitioned across a Kafka consumer group, keep validation CPU-bound and per-partition so a slow tenant cannot head-of-line block others, and treat the dead-letter queue as its own partition with independent consumer isolation so triage of rejected payloads never competes with live validation throughput. A single worker should sustain low-thousands of validations per second; if it does not, the bottleneck is almost always per-payload schema recompilation or synchronous audit-log I/O — batch the audit writes.

Testing & Compliance Verification #

Treat the validator as the artifact a regulator will replay. Maintain a test payload matrix that pairs each jurisdiction with each request_type and each named edge case, and assert the exact rejection path — not merely that a ValidationError was raised.

import pytest
from pydantic import ValidationError


@pytest.mark.parametrize(
    "field,value,expected_loc",
    [
        ("subject_id", "abc", "subject_id"),          # fails pattern
        ("jurisdiction", "USA", "jurisdiction"),      # 3 chars, not ISO-3166 alpha-2
        ("request_type", "export", "request_type"),   # not in enum
    ],
)
def test_contract_rejects_with_exact_path(field, value, expected_loc):
    payload = {
        "subject_id": "AB12CD34",
        "request_type": "access",
        "timestamp": "2026-07-01T00:00:00+00:00",
        "jurisdiction": "DE",
    }
    payload[field] = value
    with pytest.raises(ValidationError) as exc:
        DSRIngestionContract.model_validate(payload)
    assert exc.value.errors()[0]["loc"][0] == expected_loc


def test_naive_timestamp_is_rejected():
    with pytest.raises(ValidationError):
        DSRIngestionContract.model_validate(
            {
                "subject_id": "AB12CD34",
                "request_type": "deletion",
                "timestamp": "2026-07-01T00:00:00",  # no offset
                "jurisdiction": "FR",
            }
        )

Hold out at least one regulatory region from the training/fixture set and run it only in a regression gate, so contract drift for a less-common jurisdiction is caught before deployment rather than in production. Assert that every accepted payload produces a stable payload_hash across runs — a changing hash for identical input means normalization is non-deterministic and the audit trail is not replayable.

Frequently Asked Questions #

Related #

• Cross-System Data Discovery & Sync — the parent architecture whose read-only discovery output this gate accepts or dead-letters
• Validating JSON Payloads Against DSR Schemas — step-by-step edge-case debugging and secure fallback routing for the four hardest payload shapes
• Database Connector Configuration — how native SQL types are mapped to the contract primitives validated here
• SaaS API Sync Strategies — the normalization sources whose heterogeneous payloads Phase 2 reshapes before validation
• Async Polling & Queue Management — the retry-versus-quarantine state machine that routes payloads to and from this gate