JSON Schema Validation for Drug Records

Strict JSON Schema validation is the deterministic compliance gate for every drug record entering pharmacy inventory. Before a payload reaches the perpetual ledger, triggers reconciliation, or adjusts a controlled-substance balance, it must pass structural, semantic, and regulatory verification in a single fail-closed pass. A record is treated as an unverified assertion until that pass completes — a structurally valid object with a semantically invalid NDC, an expired lot, or a misclassified schedule is still rejected. This validation boundary is the foundational control within the broader Data Ingestion & Inventory Sync Workflows architecture, and it is engineered so that compliance is encoded as a structural constraint at ingestion rather than reconstructed after the fact during a DEA or FDA inspection.

Regulatory Context & Compliance Boundaries

Validation sits at the intersection of three federal frameworks. The schema must encode each as a hard precondition before ledger commitment, not as a downstream report.

The decisive design principle is additionalProperties: false. An open schema silently accepts unknown fields, which is precisely how PHI leaks into an inventory store and how schema drift defeats an audit. Controlled-substance records are additionally routed through the DEA Schedule II–V Classification Mapping engine, and every code is normalized using the rules defined in NDC-11 vs NDC-10 Parsing Standards before it can serve as a ledger primary key.

Schema Specification & Field Map

The contract is a Draft 2020-12 schema. Inventory payloads are deliberately narrow: they describe product, quantity, and provenance, and nothing about the patient. The fields the gate depends on are mapped below.

Two rules carry most of the compliance weight. First, the NDC pattern enforces the 11-digit segmented form so the value can act as a stable join key against the FDA NDC Directory. Second, validation_metadata is injected by the gate itself — never trusted from the inbound payload — so the recorded ingested_at and schema_version are authoritative for inspection.

Deterministic Validation Workflow

Validation is a fixed-order pipeline. Cheap, high-signal rejections run first so a malicious or malformed payload is dropped before expensive structural work, and the order is identical for every source — wholesaler EDI, automated dispensing cabinet, or POS scan — so no ingestion path can bypass a check.

1. HIPAA minimization scan. The payload’s top-level keys are intersected with the PHI blocklist. Any hit is an immediate REJECTED, logged without echoing the offending value.
2. Structural validation. The record is checked against the Draft 2020-12 schema with additionalProperties: false. Missing required fields, unknown keys, enum violations, and bad date formats fail here.
3. NDC format verification. A redundant precompiled-regex check guarantees the segmented 11-digit shape even if the schema is later relaxed — defense in depth on the ledger primary key.
4. DEA schedule logic. If is_controlled_substance is true, the schedule must be a valid C-II … C-V value and chain_of_custody_id must be present; otherwise the record is rejected before it can touch a controlled balance.
5. Metadata injection & audit hashing. Only on full success does the gate stamp validation_metadata and compute the SHA-256 hash that binds the record into the append-only ledger.

State transitions are total: every record terminates in exactly one of ACCEPTED (forwarded with an audit hash) or REJECTED (routed to quarantine with a violation list). There is no partial-accept path.

Production Python Validation Engine

The implementation below is a focused, runnable validator: typed, fail-closed, with structured logging that never emits PHI and a SHA-256 hash over the canonicalized record for ledger anchoring. It is designed to drop into a containerized microservice or serverless function handling high-throughput inventory streams.

import hashlib
import json
import logging
import re
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any

from jsonschema import Draft202012Validator, FormatChecker
from jsonschema.exceptions import SchemaError, ValidationError

# Structured logging — note: no record values are interpolated into log lines,
# so a rejected payload can never leak PHI into the audit log itself.
logging.basicConfig(
    level=logging.INFO,
    format='{"ts":"%(asctime)s","level":"%(levelname)s",'
           '"svc":"pharmacy_inventory_validator","msg":"%(message)s"}',
)
logger = logging.getLogger(__name__)

COMPLIANCE_VERSION = "2026.06.0"
NDC_PATTERN = re.compile(r"^\d{5}-\d{4}-\d{2}$")  # precompiled: 11-digit segmented NDC
HIPAA_BLOCKLIST = frozenset(
    {"patient_id", "patient_name", "npi", "mrn", "dob", "address", "phone"}
)
DEA_SCHEDULES = ("C-II", "C-III", "C-IV", "C-V")

DRUG_INVENTORY_SCHEMA: dict[str, Any] = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "type": "object",
    "additionalProperties": False,  # blocks PHI injection and schema drift
    "required": [
        "ndc", "lot_number", "expiration_date", "quantity", "unit_of_measure",
        "is_controlled_substance", "source_system_id",
    ],
    "properties": {
        "ndc": {"type": "string", "pattern": NDC_PATTERN.pattern},
        "lot_number": {"type": "string", "minLength": 1},
        "expiration_date": {"type": "string", "format": "date"},
        "quantity": {"type": "number", "minimum": 0},
        "unit_of_measure": {"type": "string", "enum": ["EA", "ML", "GM", "TAB", "CAP"]},
        "is_controlled_substance": {"type": "boolean"},
        "schedule": {"type": "string", "enum": list(DEA_SCHEDULES)},
        "chain_of_custody_id": {"type": "string", "minLength": 1},
        "source_system_id": {"type": "string", "minLength": 1},
    },
}


@dataclass(frozen=True)
class ComplianceResult:
    """Immutable outcome of one validation pass — safe to forward to the ledger."""
    accepted: bool
    status: str
    violations: tuple[str, ...] = ()
    audit_hash: str | None = None
    metadata: dict[str, Any] | None = None


class PharmacySchemaValidator:
    def __init__(self, schema: dict[str, Any]) -> None:
        try:
            self._validator = Draft202012Validator(schema, format_checker=FormatChecker())
        except SchemaError as exc:  # fail at boot, never at request time
            logger.critical("Invalid schema configuration: %s", exc)
            raise

    @staticmethod
    def _audit_hash(record: dict[str, Any]) -> str:
        """SHA-256 over the canonical JSON form binds the record into the ledger chain."""
        canonical = json.dumps(record, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def validate(self, payload: dict[str, Any]) -> ComplianceResult:
        # 1. HIPAA data minimization — cheapest, highest-signal rejection first.
        phi = HIPAA_BLOCKLIST & payload.keys()
        if phi:
            logger.warning("HIPAA blocklist hit (%d key(s)); payload quarantined", len(phi))
            return ComplianceResult(False, "REJECTED",
                                    tuple(f"HIPAA_BLOCKED_KEY:{k}" for k in sorted(phi)))

        # 2. Structural validation — additionalProperties:false enforces the closed contract.
        try:
            self._validator.validate(payload)
        except ValidationError as exc:
            logger.error("Schema validation failed at %s", list(exc.absolute_path) or "root")
            return ComplianceResult(False, "REJECTED", (f"SCHEMA_ERROR:{exc.message}",))

        # 3. Redundant NDC check — defense in depth on the ledger primary key.
        if not NDC_PATTERN.match(payload["ndc"]):
            return ComplianceResult(False, "REJECTED", ("NDC_INVALID_FORMAT",))

        # 4. DEA schedule logic — controlled items need a valid schedule + custody trail.
        if payload["is_controlled_substance"]:
            if payload.get("schedule") not in DEA_SCHEDULES:
                return ComplianceResult(False, "REJECTED", ("DEA_SCHEDULE_INVALID",))
            if not payload.get("chain_of_custody_id"):
                return ComplianceResult(False, "REJECTED", ("DEA_CUSTODY_MISSING",))

        # 5. Authoritative metadata + audit hash — never trusted from the inbound payload.
        metadata = {
            "ingested_at": datetime.now(timezone.utc).isoformat(),
            "schema_version": COMPLIANCE_VERSION,
            "compliance_framework": "FDA_DSCSA|DEA_21CFR1304|HIPAA",
        }
        sealed = {**payload, "validation_metadata": metadata}
        audit_hash = self._audit_hash(sealed)
        logger.info("Record accepted; audit_hash=%s", audit_hash)
        return ComplianceResult(True, "ACCEPTED", audit_hash=audit_hash, metadata=metadata)


if __name__ == "__main__":
    validator = PharmacySchemaValidator(DRUG_INVENTORY_SCHEMA)
    sample = {
        "ndc": "00123-4567-89",
        "lot_number": "LOT-2026-X9",
        "expiration_date": "2027-12-31",
        "quantity": 150,
        "unit_of_measure": "TAB",
        "is_controlled_substance": True,
        "schedule": "C-III",
        "chain_of_custody_id": "CUST-8842-991",
        "source_system_id": "WHS-EDI-846-NODE-04",
    }
    result = validator.validate(sample)
    print(json.dumps(result.__dict__, indent=2, default=str))

The engine uses Draft202012Validator for strict structural compliance, scans for PHI before any other work, and produces a frozen ComplianceResult so a downstream consumer cannot mutate the verdict. The deeper schema-design rationale — handler attribution, schedule cross-checks, and quantity-unit normalization — is covered in depth in the companion recipe on validating drug inventory JSON schemas.

Compliance Mapping & Audit Boundaries

A passing record does not simply get written; it gets anchored. The audit_hash returned by the gate is the link between validation and the append-only ledger: each ledger entry hashes its predecessor, so any later edit to a committed drug record breaks the chain and is detectable on audit. This satisfies the tamper-evident, complete-and-accurate retention standard of 21 CFR § 1304.11, and corrections are recorded as forward-chained entries rather than in-place edits.

Access to the validation layer and the ledger it feeds is governed by role-based controls. Ingestion service accounts may submit and read their own verdicts; only the audit role may read the full chain; and no role may delete. Every validation event — accepted and rejected alike — is written to an immutable store under WORM (Write Once, Read Many) retention, with the rejected-payload body held only in an encrypted quarantine, never in the searchable audit log. Network segmentation keeps the validation layer, the perpetual ledger, and any clinical dispensing system on separate trust zones so an inventory payload can never co-mingle with PHI.

Error Handling & Offline Resilience

The gate distinguishes two failure classes, and the distinction is itself a compliance requirement: a transient failure must never be recorded as a compliance rejection.

• Permanent rejections — invalid NDC, missing lot, PHI leakage, or bad schedule logic — are routed to a quarantine queue for manual compliance review, with an automated alert to the pharmacy operations and security teams. These are terminal; the payload is never retried as-is.
• Transient failures — a database timeout while persisting metadata, or an NDC Directory lookup that times out — trigger exponential backoff keyed by an idempotency token so a retry cannot double-post the same drug record.
• Offline windows — when an upstream reference (the FDA NDC Directory or a PDMP) is unreachable, the record is validated against a cached snapshot and flagged for deferred verification, with the original payload preserved in a sealed staging queue per the Fallback Routing for Offline Sync strategy. This keeps perpetual counts accurate without emitting premature diversion alerts, and re-validates automatically once connectivity returns.

Idempotent retry, dead-letter handling, and backoff policy for the queue that sits behind this gate are detailed in Error Handling & Retry Mechanisms.

Downstream Integration

Validation is the front door for several ingestion paths and the precondition for every subsystem that acts on inventory state. Because the gate is shared, all of them inherit identical compliance guarantees:

1. EDI reconciliation — wholesaler feeds normalized by the EDI 852 & 846 Parsing Pipelines are validated here before their quantities become perpetual-inventory deltas; a structurally valid X12 segment with an invalid drug record is still rejected at this boundary.
2. Real-time scan decrements — sub-second events from Barcode Scan Log Routing Logic pass through the same validator, so high-volume nightly EDI syncs and live POS decrements can never diverge in their schema enforcement.
3. Async ledger commitment — accepted records, each carrying its audit hash, are queued for idempotent upsert by Async Batch Processing for Inventory Updates, which owns the dead-letter and idempotency-key handling for the write path.

By enforcing a closed schema, scanning for PHI before structural work, and binding every accepted record to a SHA-256 audit hash, the validation gate converts heterogeneous inbound payloads into a uniform, compliant, ledger-ready inventory stream — eliminating manual reconciliation overhead while satisfying DEA, FDA, and HIPAA audit requirements at scale.

Frequently Asked Questions

Related

• Data Ingestion & Inventory Sync Workflows — parent architecture this validation gate operates within
• Validating drug inventory JSON schemas — the deeper schema-design recipe for this gate
• EDI 852 & 846 Parsing Pipelines — upstream feed normalized before validation
• Async Batch Processing for Inventory Updates — idempotent ledger commitment downstream
• Error Handling & Retry Mechanisms — quarantine, dead-letter, and backoff behind the gate
• DEA Schedule II–V Classification Mapping — schedule logic the controlled-substance check relies on