Validating drug inventory JSON schemas

A copy-paste Python validator that aggregates every DEA/FDA schema violation at once, runs custom NDC and schedule format checks, and emits a tamper-evident rejection record before any ledger write.

The default way to validate a payload with the jsonschema library — calling .validate() — raises on the first error and stops. For a controlled-substance record that is exactly the wrong behaviour: a wholesaler manifest that arrives with a malformed ndc, a missing lot_number, and a Schedule II quantity in the wrong unit gets rejected for one fault, re-sent, rejected for the next, and re-sent again. The reviewer never sees the whole picture, the retry loop burns through a retry budget, and 21 CFR § 1304.21 recordkeeping is left with a half-validated payload bouncing between queues. This page solves that specific problem: a deterministic validator that collects every violation in one pass, layers DEA- and FDA-specific format checks on top of the structural schema, and binds each rejection to a SHA-256 audit hash before anything touches the perpetual ledger. It sits under the parent topic JSON Schema Validation for Drug Records within the broader Data Ingestion & Inventory Sync Workflows subsystem.

The validator has to answer four questions for every payload in a single, repeatable evaluation: is the structure well-formed, are the regulated fields (ndc, dea_schedule, facility_dea_number, expiration_date) individually valid, does the record carry protected health information it must not, and — if any of those fail — what is the complete, ordered list of faults plus a hash that proves what was rejected. When all of that resolves in one place, a bad payload becomes a single quarantined record with a full violation list, not a stream of one-error-at-a-time retries.

Single-pass inventory validation: collect every fault, then branch on an empty violation list An inventory payload is hashed once with SHA-256 over its canonical bytes, then run through one evaluation pass that never raises on the first error. The pass has three stages that all append to a shared list: a HIPAA/PHI key scan that runs first and never logs values, a structural schema pass using Draft202012Validator.iter_errors() that yields every schema fault with no early exit, and a custom format pass that checks ndc-11, dea-number, and future-date independently. The stages aggregate into one ordered, complete violation list. A decision tests whether the list is empty: an empty list yields an ACCEPTED verdict routed to the append-only perpetual ledger with the audit hash bound to the verdict, while one or more faults yields a REJECTED verdict routed to the quarantine queue with the audit hash plus the full violation list. Inventory payload (dict) audit_hash = SHA-256(canonical bytes) Single evaluation pass — collect every fault, never raise on the first 1 · HIPAA / PHI key scan HIPAA_BLOCKLIST ∩ payload.keys() — runs first · values never logged 2 · Structural schema pass Draft202012Validator.iter_errors() — yields every schema fault, no early exit 3 · Custom format pass ndc-11 · dea-number · future-date — each checked independently aggregate violations[] — one ordered, complete list HIPAA_BLOCKED_KEY:patient_name SCHEMA:ndc:'123-45-6' is not a 'ndc-11' SCHEMA:dea_schedule:'I' is not one of [...] len == 0 ? empty → accept ≥ 1 fault → reject ACCEPTED → perpetual ledger (append-only) audit_hash bound to verdict REJECTED → quarantine queue audit_hash + full violation list

Prerequisites & environment

  • Python 3.11+ — the implementation uses dataclasses(frozen=True), enum.Enum, and X | None union syntax.
  • jsonschema 4.18+ for the Draft202012Validator, iter_errors, and the FormatChecker extension API. Everything else (re, hashlib, json, logging, datetime) is standard library. Add pytest only for the test block.
  • Regulatory context you should already hold: 21 CFR § 1304.21 (records of receipt and disposition must identify each substance exactly), 21 CFR § 1304.22 (record retention and content), 21 CFR § 1306.21 for handling, and HIPAA § 164.312(b) (audit controls for electronic systems). Inventory payloads are not dispensing records, so they must never carry patient identifiers — that data-minimization boundary is enforced in code below.
  • The ndc field is assumed to already be canonical NDC-11. This validator checks shape, not normalization; the rules that collapse 4-4-2, 5-3-2, and 5-4-1 inputs into the canonical form live in NDC-11 vs NDC-10 Parsing Standards, with a copy-paste implementation in NDC parsing regex patterns for Python. Run that parser before this validator.

The fields the schema regulates and the statute each one answers to:

Field Constraint Compliance basis
ndc NDC-11 ^\d{5}-\d{4}-\d{2}$ FDA NDC Directory format; 21 CFR § 207.33
dea_schedule enum II, III, IV, V (Schedule I excluded) 21 CFR § 1308; 21 CFR § 1304.21
lot_number alphanumeric, 3–20 chars DSCSA traceability; 21 CFR § 1304.22
expiration_date ISO 8601, strictly after ingestion time prevents expired stock entering dispensing
quantity / uom non-negative number + unit enum 21 CFR § 1304.22 exact-quantity records
transaction_type enum receipt, dispense, adjustment, destruction, transfer maps to Form 222 / ARCOS categories
facility_dea_number ^[A-Z]{2}\d{7}$ registrant attribution; 21 CFR § 1301.12

Implementation: a single-pass validator that aggregates every violation

The core change from the naive pattern is iter_errors instead of validate: it yields all structural errors rather than raising on the first. Custom format checks (ndc-11, dea-number, future-date) are registered on a FormatChecker so they run inside the same evaluation and surface through the same error stream. The HIPAA key scan runs first because a payload carrying PHI must be rejected outright, never logged with its contents. The result is a frozen ValidationOutcome whose audit_hash binds the raw payload to its verdict.

python
import json
import re
import hashlib
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from jsonschema import Draft202012Validator, FormatChecker

logger = logging.getLogger("pharmacy.inventory.validation")

# PHI keys that must never appear in an inventory payload (HIPAA § 164.312 data
# minimization). Inventory is not a dispensing record; reject, do not log values.
HIPAA_BLOCKLIST = frozenset(
    {"patient_id", "patient_name", "npi", "mrn", "dob", "address", "phone", "rx_number"}
)

# Custom format checker shared by the validator so NDC/DEA/date rules surface
# through the same iter_errors stream as the structural constraints.
pharmacy_formats = FormatChecker()


@pharmacy_formats.checks("ndc-11")
def _is_ndc_11(value: object) -> bool:
    return isinstance(value, str) and bool(re.fullmatch(r"\d{5}-\d{4}-\d{2}", value))


@pharmacy_formats.checks("dea-number")
def _is_dea_number(value: object) -> bool:
    return isinstance(value, str) and bool(re.fullmatch(r"[A-Z]{2}\d{7}", value))


@pharmacy_formats.checks("future-date")
def _is_future_date(value: object) -> bool:
    if not isinstance(value, str):
        return False
    try:
        return datetime.fromisoformat(value).date() > datetime.now(timezone.utc).date()
    except ValueError:
        return False


INVENTORY_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    "$id": "pharma/inventory/v1",
    "type": "object",
    "additionalProperties": False,  # reject vendor drift, never silently absorb it
    "required": [
        "ndc", "dea_schedule", "lot_number", "expiration_date",
        "quantity", "uom", "transaction_type", "facility_dea_number",
    ],
    "properties": {
        "ndc": {"type": "string", "format": "ndc-11"},
        "dea_schedule": {"type": "string", "enum": ["II", "III", "IV", "V"]},
        "lot_number": {"type": "string", "minLength": 3, "maxLength": 20,
                       "pattern": "^[A-Za-z0-9-]+$"},
        "expiration_date": {"type": "string", "format": "future-date"},
        "quantity": {"type": "number", "minimum": 0},
        "uom": {"type": "string", "enum": ["EA", "ML", "GM", "TAB", "CAP", "VIAL"]},
        "transaction_type": {"type": "string", "enum": [
            "receipt", "dispense", "adjustment", "destruction", "transfer"]},
        "facility_dea_number": {"type": "string", "format": "dea-number"},
    },
}

_VALIDATOR = Draft202012Validator(INVENTORY_SCHEMA, format_checker=pharmacy_formats)


class Verdict(str, Enum):
    ACCEPTED = "ACCEPTED"
    REJECTED = "REJECTED"


@dataclass(frozen=True)
class ValidationOutcome:
    verdict: Verdict
    violations: tuple[str, ...]      # ordered, complete list — never just the first
    audit_hash: str                  # SHA-256 over the canonical raw payload
    evaluated_at: str
    flags: dict[str, bool] = field(default_factory=dict)


def _audit_hash(payload: dict) -> str:
    """Tamper-evident hash of the exact bytes evaluated, independent of verdict."""
    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def validate_inventory_record(payload: dict) -> ValidationOutcome:
    """
    Validate one drug-inventory payload in a single pass.

    Collects every structural and format violation (not just the first), plus
    any HIPAA blocklist hits, into one ordered list. Binds the result to a
    SHA-256 hash of the exact payload so a quarantined record is provably the
    one that was rejected (21 CFR § 1304.21 identity; HIPAA § 164.312(b) audit).
    """
    audit_hash = _audit_hash(payload)
    violations: list[str] = []

    # 1. HIPAA data minimization runs first; never emit blocked values to logs.
    phi_hits = sorted(HIPAA_BLOCKLIST & payload.keys())
    violations.extend(f"HIPAA_BLOCKED_KEY:{k}" for k in phi_hits)

    # 2. Structural + custom-format pass — iter_errors yields ALL faults at once.
    for err in sorted(_VALIDATOR.iter_errors(payload), key=lambda e: list(e.path)):
        path = "/".join(str(p) for p in err.path) or "<root>"
        violations.append(f"SCHEMA:{path}:{err.message}")

    verdict = Verdict.REJECTED if violations else Verdict.ACCEPTED
    evaluated_at = datetime.now(timezone.utc).isoformat()

    # 3. PHI-free structured audit line: identifiers, verdict, and hash only.
    if verdict is Verdict.REJECTED:
        logger.warning(
            "INVENTORY_REJECTED audit_hash=%s violations=%d schedule=%s",
            audit_hash, len(violations), payload.get("dea_schedule", "?"),
        )
    else:
        logger.info(
            "INVENTORY_ACCEPTED audit_hash=%s ndc=%s txn=%s",
            audit_hash, payload.get("ndc"), payload.get("transaction_type"),
        )

    return ValidationOutcome(
        verdict=verdict,
        violations=tuple(violations),
        audit_hash=audit_hash,
        evaluated_at=evaluated_at,
        flags={"had_phi": bool(phi_hits)},
    )

The outcome is frozen, so once a payload is judged the binding of its bytes to verdict and hash cannot be edited in place — the property an append-only audit boundary depends on. A record whose verdict is ACCEPTED is safe to route into the async batch update path; a REJECTED outcome carries the full violation list straight to the quarantine queue.

Verification & testing

Two properties must be proven: that a multi-fault payload returns all its faults in one call, and that the audit hash is stable and bound to the exact bytes evaluated. The block below asserts both, plus the HIPAA short-circuit and the future-date rule.

python
import pytest

VALID = {
    "ndc": "00123-4567-89",
    "dea_schedule": "II",
    "lot_number": "LOT-2026-X9",
    "expiration_date": "2030-12-31",
    "quantity": 150,
    "uom": "TAB",
    "transaction_type": "receipt",
    "facility_dea_number": "AB1234567",
}


def test_accepts_a_clean_schedule_ii_record():
    out = validate_inventory_record(VALID)
    assert out.verdict is Verdict.ACCEPTED
    assert out.violations == ()


def test_collects_every_violation_in_one_pass():
    bad = {**VALID, "ndc": "123-45-6", "dea_schedule": "I",
           "facility_dea_number": "bad", "expiration_date": "1999-01-01"}
    out = validate_inventory_record(bad)
    assert out.verdict is Verdict.REJECTED
    # Four independent faults must all surface from a single call.
    joined = " ".join(out.violations)
    assert "ndc" in joined and "dea_schedule" in joined
    assert "facility_dea_number" in joined and "expiration_date" in joined
    assert len(out.violations) >= 4


def test_phi_is_rejected_and_never_silently_passed():
    out = validate_inventory_record({**VALID, "patient_name": "redacted"})
    assert out.verdict is Verdict.REJECTED
    assert any(v.startswith("HIPAA_BLOCKED_KEY:patient_name") for v in out.violations)
    assert out.flags["had_phi"] is True


def test_audit_hash_is_stable_and_byte_bound():
    out = validate_inventory_record(VALID)
    assert out.audit_hash == _audit_hash(VALID)              # reproducible
    assert out.audit_hash != _audit_hash({**VALID, "quantity": 151})

To validate the compliance trail rather than the logic, capture the emitted log line for a rejection and confirm it carries the hash and a fault count but no payload values or PHI:

text
WARNING pharmacy.inventory.validation INVENTORY_REJECTED \
    audit_hash=8c1d…b07f violations=4 schedule=I

Re-running _audit_hash(stored_payload) against a quarantined record and comparing it to the persisted audit_hash is the inspection check: a mismatch proves the payload was altered after it was judged, which is exactly the tamper signal a 21 CFR § 1304.22 retention audit is meant to surface.

Gotchas & compliance pitfalls

  • validate() masks faults; iter_errors() reveals them. The single biggest mistake is keeping the first-error-wins call in a regulated pipeline. A reviewer needs the complete list to disposition a manifest in one ticket, and the retry layer should not re-evaluate the same payload five times to discover five separate faults.
  • additionalProperties: false is non-negotiable here. Setting it to true silently absorbs vendor-specific keys, which is how PHI or undocumented fields leak past the gate. The closed schema is what makes the EDI 852/846 parsing pipeline output verifiable.
  • format is advisory unless you attach a FormatChecker. Without passing format_checker=pharmacy_formats, the ndc-11, dea-number, and future-date formats are ignored entirely and malformed values sail through. Always wire the checker into the validator instance.
  • Schedule I must be rejected, not mapped. Schedule I substances are not part of standard pharmacy dispensing inventory; a payload claiming dea_schedule: "I" is malformed by definition. The enum enforces this, and the DEA Schedule II-V Classification Mapping engine downstream assumes the value is already constrained to II–V.
  • additionalProperties only blocks unknown keys — it cannot detect a PHI key that you forgot to blocklist. Keep HIPAA_BLOCKLIST and the schema in sync, and prefer rejecting unknown keys outright over enumerating every possible PHI field.
  • Naive date validation accepts expired stock. format: "date" only checks ISO shape; it will happily pass 1999-01-01. The future-date checker compares against the ingestion date so expired lots cannot enter a dispensing queue.
  • Offline ingestion still validates on replay. A payload buffered during a disconnect must be re-run through validate_inventory_record on reconnect, not trusted because it was already queued. The deferred-validation policy is owned by Fallback routing for offline sync; the rejection rules on replay are identical to the online path.

Frequently Asked Questions

Why aggregate every error instead of failing fast?

A controlled-substance payload that fails fast gets rejected one fault at a time, so a manifest with four problems takes four ingestion cycles to fully diagnose. A compliance reviewer needs the complete, ordered violation list to disposition the record in a single quarantine ticket, and the error handling and retry layer should not waste a retry budget rediscovering known faults.

Does this validator normalize the NDC?

No. It checks that ndc already matches the canonical NDC-11 shape and rejects anything else. Normalization — collapsing 4-4-2, 5-3-2, and 5-4-1 inputs into 5-4-2 — happens earlier, per NDC parsing regex patterns for Python. Keeping the two concerns separate means a barcode-format change cannot weaken the validation gate.

How does the audit hash help during a DEA inspection?

The audit_hash is a SHA-256 over the exact bytes that were evaluated, stored alongside the verdict. Re-hashing a retained record and comparing it to the stored value proves whether the payload was altered after judgment — the tamper-evidence that 21 CFR § 1304.22 record integrity expects, without storing any patient data.

Why does the HIPAA check run before structural validation?

Because a payload carrying patient identifiers must be rejected and quarantined without ever logging its contents. Running the blocklist scan first guarantees PHI never reaches a log line or a downstream error message, keeping HIPAA § 164.312(b) audit controls separate from inventory data.