Pydantic Models for Drug-Record Validation

A drug record is an untrusted assertion until it has been validated, and the cheapest place to reject a bad one is the ingestion boundary — before it can reach the controlled-substance ledger, trigger a reconciliation, or move a Schedule II balance. Pydantic v2 turns that boundary into typed, declarative code: an NDC that is not 11 digits, a schedule outside C-II…C-V, a negative quantity, or an expired lot each raises a ValidationError and routes the payload to quarantine instead of the ledger. This recipe builds those models, and it operates within the JSON Schema Validation for Drug Records gate, giving the same fail-closed guarantees as a hand-written schema with far less boilerplate and stronger typing.

Pydantic earns its place here because the constraints a drug record needs are precisely the ones Pydantic expresses natively: field validators for the NDC shape, an Enum for the DEA schedule, numeric bounds for quantity, and cross-field checks for the controlled-substance custody rule. The trap is that Pydantic is helpful by default — it will coerce "12" into 12, accept a stray field if you let it, and echo the offending value into an error message that then lands in a log. A model that protects a DEA record has to switch those conveniences off deliberately.

Problem framing & prerequisites

The record entering this model is a candidate, not a fact. It may carry an NDC padded in the wrong configuration, a schedule string a vendor invented, a quantity that arrived as a float, or — worst — a stray patient_id a misconfigured upstream attached. The model’s job is to accept only records that satisfy every structural and regulatory constraint and to reject the rest into quarantine, without ever letting a coercion quietly “fix” a value into something plausible-but-wrong.

Environment:

• Python 3.11+ and Pydantic v2 (pip install "pydantic>=2.6"). The v2 API used below — field_validator, model_validator, ConfigDict, model_config — differs from v1; this recipe is v2-only.
• Standard library hashlib, logging, datetime, enum for the audit seal and structured logging. Add pytest for the verification block.
• Regulatory grounding: 21 CFR § 1304.21 (complete-and-accurate records with exact product identity), 21 CFR § 1304.22 (exact-unit logging and custody trail), DSCSA 21 USC § 360eee (lot/expiry traceability), and HIPAA 45 CFR § 164.514(b) (data minimization — no PHI in an inventory record). The NDC must already be normalized to canonical 11-digit form using the NDC-11 vs NDC-10 Parsing Standards rules before the model validates its shape, and the schedule value is the one produced by the DEA Schedule II–V Classification Mapping engine.

Implementation: a strict, fail-closed drug-record model

The model turns every convenience off: strict=True blocks silent coercion, extra="forbid" rejects unknown keys (the HIPAA boundary), and a field_validator enforces the NDC shape without re-padding it — normalization is upstream. A model_validator handles the cross-field rule that a controlled item must carry both a valid schedule and a chain-of-custody id. Error messages are scrubbed so a rejected payload never leaks a value into a log.

import hashlib
import json
import logging
import re
from datetime import date, datetime, timezone
from decimal import Decimal
from enum import Enum

from pydantic import (
    BaseModel, ConfigDict, Field, field_validator, model_validator,
)
from pydantic import ValidationError  # re-exported for callers

logger = logging.getLogger("pharmacy.drug_record")

# NDC must already be canonical 11-digit segmented — this checks shape, not padding.
_NDC_11 = re.compile(r"^\d{5}-\d{4}-\d{2}$")
_PHI_KEYS = frozenset({"patient_id", "patient_name", "npi", "mrn", "dob", "address"})


class DEASchedule(str, Enum):
    C_II = "C-II"
    C_III = "C-III"
    C_IV = "C-IV"
    C_V = "C-V"


class UnitOfMeasure(str, Enum):
    EA = "EA"
    ML = "ML"
    GM = "GM"
    TAB = "TAB"
    CAP = "CAP"


class DrugRecord(BaseModel):
    # strict=True: no "12" -> 12 coercion; extra="forbid": unknown keys (PHI) rejected.
    model_config = ConfigDict(strict=True, extra="forbid", frozen=True)

    ndc: str
    lot_number: str = Field(min_length=1)
    expiration_date: date
    quantity: Decimal = Field(gt=0)                 # positive on-hand movement
    unit_of_measure: UnitOfMeasure
    is_controlled_substance: bool
    schedule: DEASchedule | None = None
    chain_of_custody_id: str | None = None
    source_system_id: str = Field(min_length=1)

    @field_validator("ndc")
    @classmethod
    def _ndc_is_canonical_11(cls, v: str) -> str:
        # Reject shape only; do NOT re-pad here — normalization is upstream.
        if not _NDC_11.match(v):
            raise ValueError("NDC must be canonical 11-digit 5-4-2 form")
        return v

    @field_validator("expiration_date")
    @classmethod
    def _not_expired(cls, v: date) -> date:
        if v < datetime.now(timezone.utc).date():
            raise ValueError("expiration_date is in the past")
        return v

    @model_validator(mode="after")
    def _controlled_needs_schedule_and_custody(self) -> "DrugRecord":
        # 21 CFR § 1304.22: a controlled item requires a schedule + custody trail.
        if self.is_controlled_substance:
            if self.schedule is None:
                raise ValueError("controlled substance requires a DEA schedule")
            if not self.chain_of_custody_id:
                raise ValueError("controlled substance requires chain_of_custody_id")
        return self

    def audit_hash(self) -> str:
        """SHA-256 over the canonical record binds it into the ledger chain."""
        canonical = json.dumps(self.model_dump(mode="json"),
                               sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def validate_drug_record(payload: dict) -> DrugRecord:
    """Validate one payload, quarantining malformed input before the ledger.

    Raises ValidationError (routed to quarantine by the caller). The error is
    logged by field LOCATION only — never by value — so PHI in a bad payload
    cannot reach the audit log (HIPAA § 164.514(b)).
    """
    # Defence in depth: reject PHI keys before Pydantic even sees the payload.
    leaked = _PHI_KEYS & payload.keys()
    if leaked:
        logger.warning("DRUG_RECORD_PHI_BLOCKED keys=%d", len(leaked))
        raise ValueError(f"PHI keys present: {len(leaked)}")

    try:
        record = DrugRecord.model_validate(payload)
    except ValidationError as exc:
        # Log the failing field paths, not the offending values.
        locations = [".".join(map(str, e["loc"])) for e in exc.errors()]
        logger.error("DRUG_RECORD_REJECTED fields=%s", locations)
        raise

    logger.info("DRUG_RECORD_ACCEPTED ndc=%s audit_hash=%s",
                record.ndc, record.audit_hash())
    return record

Two decisions carry the compliance weight. strict=True means a quantity that arrives as the string "150" is rejected rather than coerced to 150 — because a system that silently coerces types will just as silently coerce a malformed value into a plausible one, and a plausible-but-wrong controlled-substance quantity is exactly the error an audit is meant to catch. And the except ValidationError handler logs e["loc"] (the field path) but never e["input"] (the value), so a payload that carried PHI or a sensitive quantity cannot bleed into the audit log. A rejected record is handed to the quarantine described in Dead-Letter Queue Design for Inventory Events, where its original payload is held under encryption rather than logged.

Verification & testing

The suite must prove both directions: a well-formed record validates and produces a stable hash, and each class of malformed record raises ValidationError — coercion off, extra fields rejected, past expiry caught, and the controlled-substance custody rule enforced.

import pytest
from pydantic import ValidationError

def _base(**over):
    p = {
        "ndc": "00093-0058-01", "lot_number": "L-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",
        "source_system_id": "WHS-EDI-846-04",
    }
    p.update(over)
    return p

def test_valid_record_passes_and_hashes_stably():
    rec = validate_drug_record(_base())
    assert rec.ndc == "00093-0058-01"
    assert rec.audit_hash() == rec.audit_hash()          # deterministic

def test_bad_ndc_shape_raises():
    with pytest.raises(ValidationError):
        validate_drug_record(_base(ndc="93-58-1"))       # not canonical 5-4-2

def test_string_quantity_is_not_coerced():
    with pytest.raises(ValidationError):
        validate_drug_record(_base(quantity="150"))      # strict mode: no coercion

def test_extra_field_is_forbidden():
    with pytest.raises(ValidationError):
        validate_drug_record(_base(reorder_hint="soon"))  # extra="forbid"

def test_past_expiry_raises():
    with pytest.raises(ValidationError):
        validate_drug_record(_base(expiration_date="2020-01-01"))

def test_controlled_without_custody_raises():
    with pytest.raises(ValidationError):
        validate_drug_record(_base(chain_of_custody_id=None))

def test_phi_key_is_blocked_before_model():
    with pytest.raises(ValueError):
        validate_drug_record(_base(mrn="123456"))        # never reaches the model

A representative accepted-record log line carries the NDC and hash but no lot-linked or patient data beyond the product identity itself:

INFO pharmacy.drug_record DRUG_RECORD_ACCEPTED ndc=00093-0058-01 \
     audit_hash=a4f1c9...77e2

A rejection logs only the failing field paths, so an inspector can see what failed without the payload ever exposing a value:

ERROR pharmacy.drug_record DRUG_RECORD_REJECTED fields=['quantity', 'schedule']

Gotchas & compliance pitfalls

• Coercion surprises. Without strict=True, Pydantic v2 will happily turn "150" into 150, 1 into True, and "C-III " (trailing space) handling varies. A coerced value is an unaudited transformation of a DEA field. Turn strict mode on and make every acceptable type explicit.
• model_config strict mode is per-model, not global. Setting strict=True on one model does not propagate to nested models or to model_validate calls elsewhere. Declare it in each model’s ConfigDict, and prefer extra="forbid" on every model that touches inventory so unknown keys are always rejected.
• Validating the NDC without normalizing first. The field_validator here checks shape, not padding. If a 10-digit or wrongly-padded NDC reaches the model, it is rejected — correctly — but the fix is to normalize upstream, not to loosen the pattern. Never re-pad inside the model; that hides the ambiguity the parsing rules exist to surface.
• PHI leaking into error messages. Pydantic’s default ValidationError includes the offending input value in e["input"]. Logging the raw error, or str(exc), can dump a patient_id or a sensitive quantity straight into the audit store. Log e["loc"] only, and block PHI keys before the model runs.
• Decimal versus float for quantity. A float quantity re-introduces the formatting non-determinism that breaks downstream hashing. Declare quantity: Decimal and keep it a Decimal through the audit hash.
• Trusting client-supplied metadata. Fields like an ingestion timestamp or schema version must be stamped by the gate, never accepted from the payload — an inbound value is an unverified claim. Keep those out of the inbound model and inject them on acceptance.

Frequently Asked Questions

Related

• JSON Schema Validation for Drug Records — parent gate these models enforce at the application boundary.
• NDC-11 vs NDC-10 Parsing Standards — normalization that must run before the NDC field validator checks shape.
• DEA Schedule II–V Classification Mapping — the source of the schedule value the enum accepts.
• Dead-Letter Queue Design for Inventory Events — where a payload that raises ValidationError is quarantined.