How to map DEA schedules to inventory databases

A deterministic recipe for binding every NDC in a pharmacy inventory database to its DEA schedule — normalized, hash-chained, and resilient to offline sync — so schedule drift surfaces as a test failure, not a DEA finding.

Mapping DEA schedules to an inventory database is not a lookup column you fill once and forget — it is a control surface that drifts. A wholesaler feed pads an NDC differently, a formulary refresh adds a manufacturer’s repackaged opioid under a code no rule recognizes, a combination product gets filed under its lower-schedule adjuvant, and the gap stays invisible until an ARCOS reconciliation breaks or a state board inspector pulls the perpetual inventory. This page solves one specific problem: building a single, deterministic NDC-to-schedule binding that is normalized at ingestion, hash-chained for tamper evidence, and safe to operate while the upstream classification source is offline. It operates within the parent topic, DEA Schedule II-V Classification Mapping, and the broader Core Architecture & DEA Compliance Frameworks.

Concretely, a correct mapping answers three questions for every row, in one place, and never delegates the answer to incidental trigger or report code:

  1. Identity — what is the canonical NDC-11 for this product, regardless of how the source formatted it?
  2. Schedule — which II, III, IV, V, or non-controlled (NC) class applies, per 21 CFR § 1308, with combination products resolved to the highest constituent schedule?
  3. Provenance — which signed source asserted that schedule, when was it last verified, and can the assertion be proven unaltered?

When those three resolve deterministically, schedule misassignment becomes a unit-test failure instead of a compliance finding.

Lifecycle of one NDC-to-schedule mapping row A top-to-bottom state machine for a single mapping. A raw NDC ingests and is normalized to canonical NDC-11; an unparseable value branches right to a rejected state that logs a warning and writes no row. The normalized NDC resolves its schedule from a cryptographically signed source, then the resolver computes an audit hash and chains it to the previous chain head, and the row is persisted as the active version. When the signed source is offline, resolution detours left into an idempotent SQLite fallback queue that drains and reconciles on reconnect. A re- or de-scheduling event does not mutate the active row: it inserts a new versioned row, flips the prior row's is_active flag to FALSE, and chains the new hash, re-entering the compute-and-chain step. Raw NDC ingests Normalize → NDC-11 strip · pad to 11 digits · derive NDC-10 Resolve schedule signed source · combo → highest constituent Compute audit hash SHA-256 · chain to prior hash Persist active row is_active = TRUE · append-only Rejected log warning · no row written SQLite fallback queue idempotent · INSERT OR IGNORE drains & reconciles on reconnect Re/de-schedule event insert NEW versioned row prior row is_active = FALSE never mutate in place chain new hash → re-enters above unparseable offline reconcile

Prerequisites & environment

  • Python 3.11+ (uses dataclasses(frozen=True) and the enum.Enum value semantics shown below).
  • PostgreSQL 13+ for the canonical store (gen_random_uuid() and GENERATED ALWAYS AS ... STORED columns); SQLite 3.35+ for the local fallback cache.
  • Standard library plus requests for the resolver: hashlib, json, logging, sqlite3, enum, dataclasses.
  • A cryptographically signed NDC->schedule reference, refreshed from the DEA controlled-substance list and the FDA NDC Directory. The normalization rules that feed it are defined in NDC-11 vs NDC-10 Parsing Standards; the offline retry and reconciliation policy is owned by Fallback routing for offline sync.
  • Regulatory context you should already hold: 21 CFR § 1308.12 (Schedule II listings and combination-product rules), 21 CFR § 1304.04 (record retention, two years federally and longer in stricter states), 21 CFR § 1304.21 (recordkeeping for controlled substances), and HIPAA § 164.312(b) (audit controls). Schedule metadata must never be co-stored with patient identifiers.

Implementation: deterministic schema and resolver

Canonical store with normalization and hash lineage

The database enforces a single NDC-11 identity and derives the NDC-10 form deterministically, so wholesaler padding variants cannot create phantom duplicates. The audit_hash column makes every row self-describing: a schedule change must insert a new version, never update in place.

sql
CREATE TABLE controlled_substance_mapping (
    mapping_id   UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    ndc_11       CHAR(11) NOT NULL UNIQUE,
    ndc_10       CHAR(10) GENERATED ALWAYS AS (
        CASE
            WHEN ndc_11 ~ '^\d{11}$'
            THEN SUBSTRING(ndc_11, 2, 10)  -- canonical 5-4-2 strips the pad
            ELSE NULL
        END
    ) STORED,
    dea_schedule VARCHAR(2) NOT NULL CHECK (dea_schedule IN ('II','III','IV','V','NC')),
    source_of_truth VARCHAR(50) NOT NULL DEFAULT 'DEA_CONTROLLED_LIST',
    last_verified TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    is_active    BOOLEAN NOT NULL DEFAULT TRUE,
    audit_hash   CHAR(64) GENERATED ALWAYS AS (
        ENCODE(SHA256(CONCAT(ndc_11, dea_schedule, last_verified)::bytea), 'hex')
    ) STORED
);

CREATE INDEX idx_schedule_lookup    ON controlled_substance_mapping (dea_schedule, is_active);
CREATE INDEX idx_active_verification ON controlled_substance_mapping (last_verified DESC)
    WHERE is_active = TRUE;

Combination products (a Schedule II opioid formulated with a Schedule III/IV adjuvant) must default to the highest constituent schedule per 21 CFR § 1308.12. Enforce that at the application layer below — never let a reporting query “discover” the schedule after the fact.

The resolver: normalize, classify, hash, fall back

python
import hashlib
import json
import logging
import sqlite3
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import Optional
from urllib.parse import urljoin

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Structured audit logging per HIPAA § 164.312(b) — no PHI is ever emitted.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.FileHandler("dea_schedule_audit.log")],
)
logger = logging.getLogger("dea_schedule_mapper")


class Schedule(str, Enum):
    II = "II"
    III = "III"
    IV = "IV"
    V = "V"
    NC = "NC"  # sentinel: explicitly "not controlled", never "unknown"


# Lower ordinal = higher control. Used to resolve combination products.
_SEVERITY = {Schedule.II: 0, Schedule.III: 1, Schedule.IV: 2, Schedule.V: 3, Schedule.NC: 4}


@dataclass(frozen=True)
class NDCRecord:
    ndc_11: str
    dea_schedule: Schedule
    source: str
    verified_at: datetime
    audit_hash: str


def normalize_ndc(raw_ndc: str) -> Optional[str]:
    """Collapse any source format to canonical NDC-11. None means reject, never guess."""
    cleaned = raw_ndc.replace("-", "").strip()
    if cleaned.isdigit() and len(cleaned) == 10:
        return f"0{cleaned}"          # 10-digit -> zero-padded 11
    if cleaned.isdigit() and len(cleaned) == 11:
        return cleaned
    logger.warning("Rejected unparseable NDC: %r", raw_ndc)
    return None


def resolve_combination(schedules: list[Schedule]) -> Schedule:
    """Highest-control schedule wins, per 21 CFR § 1308.12."""
    return min(schedules, key=lambda s: _SEVERITY[s])


def compute_audit_hash(ndc: str, schedule: str, ts: datetime) -> str:
    payload = f"{ndc}|{schedule}|{ts.isoformat()}"
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


class DEAScheduleMapper:
    def __init__(self, api_base: str, fallback_db: str = "offline_schedule_cache.db"):
        self.api_base = api_base
        self.session = requests.Session()
        self.session.mount("https://", HTTPAdapter(max_retries=Retry(
            total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504],
        )))
        self.fallback_db = fallback_db
        with sqlite3.connect(self.fallback_db) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS fallback_queue (
                    ndc_11      TEXT PRIMARY KEY,
                    payload_json TEXT,
                    retry_count INTEGER DEFAULT 0,
                    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)

    def resolve_schedule(self, raw_ndc: str) -> Optional[NDCRecord]:
        ndc = normalize_ndc(raw_ndc)
        if ndc is None:
            return None
        try:
            resp = self.session.get(
                urljoin(self.api_base, f"/v1/schedule/{ndc}"), timeout=5.0,
            )
            resp.raise_for_status()
            data = resp.json()
            # A combination product returns every constituent schedule.
            constituents = [Schedule(s) for s in data.get("schedules", [data.get("schedule", "NC")])]
            schedule = resolve_combination(constituents)
            ts = datetime.now(timezone.utc)
            record = NDCRecord(
                ndc_11=ndc,
                dea_schedule=schedule,
                source=data.get("source", "DEA_CONTROLLED_LIST"),
                verified_at=ts,
                audit_hash=compute_audit_hash(ndc, schedule.value, ts),
            )
            logger.info("Resolved %s -> %s (source=%s)", ndc, schedule.value, record.source)
            return record
        except requests.RequestException as exc:
            logger.error("Resolution failed for %s, queuing for offline retry: %s", ndc, exc)
            self._enqueue_fallback(ndc)
            return None

    def _enqueue_fallback(self, ndc: str) -> None:
        """Idempotent local queue so schedule resolution never halts dispensing."""
        with sqlite3.connect(self.fallback_db) as conn:
            conn.execute(
                "INSERT OR IGNORE INTO fallback_queue (ndc_11, payload_json) VALUES (?, ?)",
                (ndc, json.dumps({"status": "pending", "reason": "upstream_unavailable"})),
            )

The schedule resolution never blocks inventory operations: when the classification source degrades, the NDC lands in an idempotent SQLite queue that the offline-sync layer drains on reconnect. Every transition is hashed before it is persisted, so the binding can be replayed and proven.

Hash-chain each accepted mapping

Tamper evidence comes from chaining, not from a single hash. Each accepted record references the previous chain head, satisfying the append-only intent of 21 CFR § 1304.04 and FDA 21 CFR Part 11 electronic-record requirements.

python
def chain_entry(prev_chain_hash: str, record: NDCRecord) -> dict:
    """Link a mapping into the append-only audit chain (Merkle-style)."""
    chain_hash = hashlib.sha256(
        f"{prev_chain_hash}|{record.audit_hash}".encode()
    ).hexdigest()
    return {
        "timestamp": record.verified_at.isoformat(),
        "ndc_11": record.ndc_11,
        "schedule": record.dea_schedule.value,
        "source": record.source,
        "row_hash": record.audit_hash,
        "prev_hash": prev_chain_hash or "genesis",
        "chain_hash": chain_hash,
    }

Persist these to WORM-grade storage (S3 Object Lock, Azure Immutable Blob). HIPAA § 164.312(c)(1) integrity controls require that audit entries cannot be altered without leaving cryptographic evidence — which the chain provides.

Verification & testing

First, prove the deterministic rules with assertions you can run in CI before the resolver ever touches the network:

python
def test_mapping_invariants():
    # Padding variants collapse to one identity.
    assert normalize_ndc("1234-5678-90") == normalize_ndc("01234567890") == "01234567890"
    assert normalize_ndc("not-an-ndc") is None        # reject, never guess

    # Combination product resolves to the most-controlled constituent.
    assert resolve_combination([Schedule.III, Schedule.II, Schedule.IV]) == Schedule.II

    # Hash is deterministic for identical inputs, divergent for any change.
    ts = datetime(2026, 6, 28, tzinfo=timezone.utc)
    h1 = compute_audit_hash("01234567890", "II", ts)
    h2 = compute_audit_hash("01234567890", "III", ts)
    assert h1 == compute_audit_hash("01234567890", "II", ts)
    assert h1 != h2

    # Chain head changes when any row changes.
    rec = NDCRecord("01234567890", Schedule.II, "DEA_CONTROLLED_LIST", ts, h1)
    assert chain_entry("genesis", rec)["chain_hash"] != \
           chain_entry("genesis", NDCRecord("01234567890", Schedule.III, "x", ts, h2))["chain_hash"]

Second, run database-side drift queries on a fixed cadence (six hours works for most operations). Target thresholds: at least 99.8% NDC-to-schedule match against the dispensing ledger, zero rows stuck at NC past a 24-hour verification window, and zero ndc_10 values that failed to derive.

sql
-- NDCs that never resolved to a real schedule and are aging out of policy.
SELECT ndc_11, last_verified FROM controlled_substance_mapping
WHERE dea_schedule = 'NC' AND is_active = TRUE
  AND last_verified < NOW() - INTERVAL '24 HOURS';

-- Dispenses that hit a product with no active mapping at all.
SELECT d.ndc_11, d.dispensed_qty
FROM dispensing_ledger d
LEFT JOIN controlled_substance_mapping m
       ON d.ndc_11 = m.ndc_11 AND m.is_active = TRUE
WHERE m.mapping_id IS NULL;

A healthy audit-log line looks like this — note that it proves an event occurred without exposing any patient identifier:

text
2026-06-28 14:02:11,407 | INFO | dea_schedule_mapper | Resolved 00093505201 -> II (source=DEA_CONTROLLED_LIST)

Gotchas & compliance pitfalls

  • NC must mean “confirmed non-controlled,” not “unknown.” Collapsing the two is the cardinal error: an unverified product silently treated as non-controlled drops out of ARCOS scope. Keep the upstream failure path in the fallback queue, distinct from a positive NC assertion.
  • Combination-product downscheduling. If a feed reports only the adjuvant’s schedule, you will file a Schedule II product as III/IV. Always request every constituent and apply resolve_combination; do not trust a single returned schedule.
  • In-place schedule updates destroy the audit trail. A UPDATE ... SET dea_schedule breaks the chain and the immutability 21 CFR § 1304.04 expects. Insert a new versioned row, flip the prior is_active to FALSE, and chain the new hash.
  • Ambiguous NDC padding. A bare 10-digit string can be a 5-4-1 or 4-4-2 source layout; blind left-padding is only valid once you know the labeler segment. Normalize through the rules in NDC-11 vs NDC-10 Parsing Standards, not with an ad-hoc zfill.
  • PHI bleed into schedule logs. It is tempting to log the prescription or patient ID alongside the NDC for debugging. Don’t — schedule metadata and dispensing PHI must stay in separate stores to keep HIPAA § 164.312(b) audit controls clean.
  • Retention shorter than the strictest state. Federal 21 CFR § 1304.04 sets two years, but several state boards require five to seven. Set the retention window to the strictest jurisdiction you operate in, applied to mappings and the hash chain.

Frequently Asked Questions

Why store NDC-11 as the key instead of NDC-10?

NDC-11 is the unambiguous canonical form; the 10-digit variants (5-4-1, 5-3-2, 4-4-2) lose the labeler-segment information needed to reconstruct it. Keying on NDC-11 and deriving NDC-10 deterministically prevents the phantom-duplicate problem where the same product appears under multiple padding layouts.

What happens to dispensing while the schedule source is offline?

Nothing stops. The resolver queues the unresolved NDC in an idempotent SQLite buffer and returns None, so the application can apply its safe-default policy (treat as controlled, route to pharmacist override) while the Fallback routing for offline sync layer reconciles on reconnect. The mapping is completed once connectivity returns, satisfying the completeness intent of 21 CFR § 1304.21 without requiring an instantaneous central write.

How do I handle a product whose schedule changes (re- or de-scheduling)?

Never mutate the existing row. Insert a new record with the new schedule and a fresh audit_hash, chain it to the prior head, and set the old row’s is_active = FALSE. The full version history is what an inspector and an ARCOS reconciliation both depend on.

Does any of this store protected health information?

No. A mapping row holds the NDC, schedule, source, verification timestamp, and hash — enough to prove the classification’s provenance without patient identifiers, keeping HIPAA § 164.312(b) audit controls separate from ePHI.