JSON Schema Validation for Drug Records
Strict JSON schema validation operates as the deterministic compliance gate for pharmacy inventory ingestion. Before any drug record enters the perpetual ledger, triggers reconciliation workflows, or
Strict JSON schema validation operates as the deterministic compliance gate for pharmacy inventory ingestion. Before any drug record enters the perpetual ledger, triggers reconciliation workflows, or updates controlled substance balances, it must pass structural, semantic, and regulatory verification. This validation boundary is the foundational control layer within Data Ingestion & Inventory Sync Workflows, ensuring that every payload aligns with FDA NDC formatting standards, DEA 21 CFR 1304 recordkeeping mandates, and HIPAA data minimization requirements.
1. Compliance Architecture & Regulatory Mapping
Validation begins with a rigid, immutable schema that maps directly to federal inventory tracking requirements. The schema must enforce three distinct compliance domains:
- FDA/DSCSA Compliance: National Drug Code (NDC) validation must strictly enforce the 11-digit segmented format (
\d{5}-\d{4}-\d{2}), mandate lot/batch traceability fields, and require ISO 8601 expiration dates. Non-compliant NDC structures or missing lot identifiers must trigger immediate rejection to prevent DSCSA traceability gaps. - DEA Controlled Substance Logging: For Schedule II–V compounds, the schema must enforce exact quantity units (e.g.,
EA,ML,GM), require chain-of-custody identifiers, and mandate explicit schedule classification. This aligns with 21 CFR Part 1304 recordkeeping requirements for controlled substances. - HIPAA Data Minimization: Inventory payloads must explicitly exclude patient identifiers, prescriber NPIs, or any Protected Health Information (PHI). Validation must reject payloads containing restricted keys before they reach downstream systems, enforcing strict data segregation between inventory management and clinical dispensing records.
The foundational structure for Validating drug inventory JSON schemas requires additionalProperties: false to prevent schema drift and enforce strict field boundaries. Every record must include a validation_metadata object capturing ingestion timestamps, source system IDs, and compliance version tags. This metadata layer ensures full audit traceability and supports regulatory inspection readiness.
2. Pipeline Integration & Routing Topology
Schema validation operates as a synchronous gate before asynchronous inventory updates. Inbound payloads typically originate from wholesale distributors, automated dispensing cabinets (ADCs), or point-of-sale (POS) terminals. When processing EDI 852 & 846 Parsing Pipelines, the EDI-to-JSON translation layer must normalize segment data into the target schema before validation executes. This normalization step strips EDI control numbers, maps LIN/QTY segments to structured JSON, and applies the compliance gate before ledger persistence.
Simultaneously, real-time dispensing events require deterministic routing. The Barcode Scan Log Routing Logic feeds scan payloads into the same validation engine, ensuring that both high-volume nightly EDI syncs and sub-second inventory decrements undergo identical compliance checks. This unified validation boundary prevents schema divergence, eliminates race conditions during concurrent inventory updates, and guarantees that all inventory state transitions are cryptographically and structurally verifiable.
3. Production-Grade Python Validation Engine
The following implementation provides a production-ready validation pipeline using jsonschema, structured logging, and explicit compliance hooks. It is designed for deployment in containerized microservices or serverless functions handling high-throughput inventory streams. The engine enforces schema boundaries, performs HIPAA key scanning, validates NDC formatting, and generates immutable compliance reports.
import json
import re
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Tuple
from jsonschema import Draft202012Validator, ValidationError, FormatChecker
from jsonschema.exceptions import SchemaError
# Structured logging configuration for audit trails
logging.basicConfig(
level=logging.INFO,
format='{"timestamp":"%(asctime)s","level":"%(levelname)s","service":"pharmacy_inventory_validator","message":"%(message)s"}'
)
logger = logging.getLogger(__name__)
# Compliance Constants
COMPLIANCE_VERSION = "2024.10.0"
NDC_REGEX = re.compile(r"^\d{5}-\d{4}-\d{2}$")
HIPAA_BLOCKLIST = {"patient_id", "patient_name", "npi", "mrn", "dob", "address", "phone"}
DEA_SCHEDULES = {"C-II", "C-III", "C-IV", "C-V"}
DRUG_INVENTORY_SCHEMA = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"additionalProperties": False,
"required": [
"ndc", "lot_number", "expiration_date", "quantity", "unit_of_measure",
"is_controlled_substance", "schedule", "source_system_id", "validation_metadata"
],
"properties": {
"ndc": {"type": "string", "pattern": NDC_REGEX.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"},
"source_system_id": {"type": "string"},
"validation_metadata": {
"type": "object",
"required": ["ingested_at", "schema_version", "compliance_framework"],
"properties": {
"ingested_at": {"type": "string", "format": "date-time"},
"schema_version": {"type": "string"},
"compliance_framework": {"type": "string", "enum": ["FDA_DSCSA", "DEA_21CFR1304", "HIPAA"]}
}
}
}
}
class PharmacySchemaValidator:
def __init__(self, schema: Dict[str, Any]):
try:
self.validator = Draft202012Validator(schema, format_checker=FormatChecker())
except SchemaError as e:
logger.critical("Invalid schema configuration: %s", str(e))
raise
def _check_hipaa_compliance(self, payload: Dict[str, Any]) -> List[str]:
"""Scan payload keys against HIPAA data minimization blocklist."""
violations = [key for key in HIPAA_BLOCKLIST if key in payload]
return violations
def _validate_ndc_structure(self, ndc: str) -> bool:
"""Enforce FDA NDC 11-digit segmented format."""
return bool(NDC_REGEX.match(ndc))
def validate_record(self, payload: Dict[str, Any]) -> Tuple[bool, Dict[str, Any]]:
"""Execute deterministic validation pipeline with compliance mapping."""
compliance_report = {
"status": "PENDING",
"timestamp": datetime.now(timezone.utc).isoformat(),
"violations": [],
"compliance_checks": {
"hipaa_minimization": "PASS",
"ndc_format": "PASS",
"schema_structure": "PASS",
"dea_schedule_logic": "PASS"
}
}
# 1. HIPAA Data Minimization Check
hipaa_violations = self._check_hipaa_compliance(payload)
if hipaa_violations:
compliance_report["status"] = "REJECTED"
compliance_report["violations"].extend([f"HIPAA_BLOCKED_KEY:{k}" for k in hipaa_violations])
compliance_report["compliance_checks"]["hipaa_minimization"] = "FAIL"
logger.warning("HIPAA violation detected in payload: %s", hipaa_violations)
return False, compliance_report
# 2. JSON Schema Structural Validation
try:
self.validator.validate(payload)
except ValidationError as e:
compliance_report["status"] = "REJECTED"
compliance_report["violations"].append(f"SCHEMA_ERROR:{e.message}")
compliance_report["compliance_checks"]["schema_structure"] = "FAIL"
logger.error("Schema validation failed: %s", e.message)
return False, compliance_report
# 3. NDC Format Verification (Redundant strict check)
if not self._validate_ndc_structure(payload.get("ndc", "")):
compliance_report["status"] = "REJECTED"
compliance_report["violations"].append("NDC_INVALID_FORMAT")
compliance_report["compliance_checks"]["ndc_format"] = "FAIL"
return False, compliance_report
# 4. DEA Schedule Logic Enforcement
if payload.get("is_controlled_substance") and payload.get("schedule") not in DEA_SCHEDULES:
compliance_report["status"] = "REJECTED"
compliance_report["violations"].append("DEA_SCHEDULE_INVALID")
compliance_report["compliance_checks"]["dea_schedule_logic"] = "FAIL"
return False, compliance_report
# 5. Compliance Metadata Injection
payload["validation_metadata"] = {
"ingested_at": compliance_report["timestamp"],
"schema_version": COMPLIANCE_VERSION,
"compliance_framework": "FDA_DSCSA|DEA_21CFR1304|HIPAA"
}
compliance_report["status"] = "ACCEPTED"
logger.info("Record validated successfully. NDC: %s | Status: %s", payload["ndc"], compliance_report["status"])
return True, compliance_report
# Usage Example for Pipeline Integration
if __name__ == "__main__":
validator = PharmacySchemaValidator(DRUG_INVENTORY_SCHEMA)
sample_payload = {
"ndc": "00123-4567-89",
"lot_number": "LOT-2024-X9",
"expiration_date": "2026-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"
}
is_valid, report = validator.validate_record(sample_payload)
print(json.dumps(report, indent=2))
This implementation aligns with best practices for Validating JSON payloads against pharmacy schema standards, ensuring deterministic rejection of malformed records while preserving audit-ready metadata. The engine uses Draft202012Validator for strict JSON Schema compliance and integrates explicit HIPAA key scanning before structural validation executes.
4. Audit Readiness & Operational Controls
Production deployment requires strict operational controls to maintain regulatory posture under DEA and FDA scrutiny. The validation engine must be integrated with centralized logging infrastructure that captures every validation event, including rejected payloads, schema drift attempts, and HIPAA blocklist triggers. Structured logs should be routed to an immutable audit store with WORM (Write Once, Read Many) retention policies.
Error handling and retry mechanisms must differentiate between transient infrastructure failures and permanent compliance rejections. Transient errors (e.g., database timeouts during metadata persistence) should trigger exponential backoff with idempotency keys. Permanent rejections (e.g., invalid NDC, missing lot numbers, or PHI leakage) must be routed to a quarantine queue for manual compliance review, with automated alerts dispatched to the pharmacy operations and security teams.
For enterprise scaling, the validation microservice should be horizontally scaled behind an API gateway with rate limiting and payload size constraints. Container orchestration must enforce network segmentation between the validation layer, the perpetual inventory ledger, and clinical dispensing systems. Regular schema version upgrades must be deployed via blue-green strategies to prevent downtime during regulatory updates.
By enforcing strict JSON schema boundaries, embedding compliance metadata at ingestion, and unifying validation across EDI and real-time POS streams, pharmacy operations achieve deterministic inventory accuracy, regulatory audit readiness, and secure data segregation across all controlled substance workflows.