Building a VRS Response Cache in Python

A dispenser scans the same saleable unit twice within ninety seconds — once at intake, once at the pharmacy counter — and both scans trigger a verification lookup against the same trading partner. Answering the second lookup from a local cache instead of a second network round trip is the difference between a sub-100ms response and a multi-second wait during a busy shift. This page is a focused build within Verification Router Service Architecture — the section of DSCSA Compliance Architecture & Standards Mapping that governs how a router answers trading-partner queries — and it assumes the router’s rate limits and retry behavior already exist. The problem here is narrower and more dangerous than ordinary response caching: cache the wrong result for even a few extra seconds and you can hand a dispenser a stale “verified” answer for a unit just flagged as suspect.

Prerequisites #

• Python 3.10+ for X | Y unions and the standard asyncio event loop used throughout.
• aiohttp or httpx for the underlying async VRS call this cache wraps.
• redis (redis-py’s redis.asyncio client), or nothing at all — the interface below works against an in-process dict for one router instance, or a shared Redis instance when the router runs as multiple replicas that must agree on cached verification state.
• Pydantic v2 for the cache-key input contract and the verification-result envelope.
• DSCSA data prerequisites — a VRS client already returning a structured result with GTIN (01), serial (21), lot (10), expiry (17), and a status of verified, not-verified, or suspect. If that circuit-broken client does not exist yet, pair this page with Implementing circuit breakers for VRS API calls, since the cache and the breaker share the same call path.

Step-by-Step Solution #

Step 1 — Build a deterministic cache key from the product identifier #

A VRS response is a fact about one specific serialized unit at one point in time, so the cache key must be derived from exactly the fields the DSCSA identifier set defines — nothing more, nothing less. Hashing the four fields into a fixed-length key keeps Redis key sizes small and avoids leaking raw serials into log lines that print cache keys.

import hashlib
from pydantic import BaseModel, field_validator

class ProductIdentifier(BaseModel):
    gtin: str     # AI (01) — 14-digit GTIN
    serial: str   # AI (21)
    lot: str      # AI (10)
    expiry: str   # AI (17), YYMMDD as received

    @field_validator("gtin")
    @classmethod
    def _gtin_len(cls, v: str) -> str:
        if len(v) != 14 or not v.isdigit():
            raise ValueError("GTIN (01) must be 14 numeric digits")
        return v

    def cache_key(self) -> str:
        # Order matters: gtin|serial|lot|expiry, never re-ordered between releases,
        # or every existing cache entry silently becomes unreachable.
        raw = f"{self.gtin}|{self.serial}|{self.lot}|{self.expiry}"
        digest = hashlib.sha256(raw.encode()).hexdigest()[:32]
        return f"vrs:verify:{digest}"

DSCSA/GS1 note: the GTIN (01) plus serial (21) alone identify the physical unit as an SGTIN, but lot (10) and expiry (17) are included in the key because a repackaging or re-labeling event can change lot or expiry association for the same serial — collapsing the key to gtin+serial would return a stale verification for a re-labeled unit.

Step 2 — Model the differentiated TTL policy explicitly #

“Verified” and “not-verified” are not symmetric outcomes. A verified result reflects a manufacturer’s commissioning record, which changes rarely, so a longer TTL is safe and reduces load on the partner endpoint. A not-verified result is frequently transient — the manufacturer’s repository has not yet ingested a just-commissioned unit — so caching it too long can make a legitimate product look permanently unverifiable to a dispenser retrying seconds later. Encode the asymmetry as data, not scattered constants.

from enum import Enum

class VerificationStatus(str, Enum):
    VERIFIED = "verified"
    NOT_VERIFIED = "not_verified"
    SUSPECT = "suspect"

# Seconds to retain a cached result per status. SUSPECT is deliberately
# absent — suspect results are never written to the cache at all.
TTL_SECONDS: dict[VerificationStatus, int] = {
    VerificationStatus.VERIFIED: 300,       # stable fact, safe to reuse briefly
    VerificationStatus.NOT_VERIFIED: 20,    # likely transient, recheck soon
}

DSCSA/GS1 note: this table is the entire compliance argument for the cache — auditors reviewing the design should be able to see, in one place, that a negative result never outlives a positive one and that suspect status has no TTL entry because it is never persisted.

Step 3 — Write the async TTL cache wrapper #

The wrapper sits directly in front of the VRS call: check the cache, call through on a miss, and only store results that are eligible for caching under the Step 2 policy. The same class works with an in-process dict or a Redis backend — swap the storage calls behind a small protocol so the calling code never changes.

import time
import asyncio
from dataclasses import dataclass
from typing import Awaitable, Callable, Protocol

@dataclass
class CachedResult:
    status: VerificationStatus
    payload: dict
    expires_at: float

class CacheBackend(Protocol):
    async def get(self, key: str) -> CachedResult | None: ...
    async def set(self, key: str, value: CachedResult, ttl: int) -> None: ...
    async def delete(self, key: str) -> None: ...

class InProcessTTLCache:
    """Single-process cache; fine for one router instance or local dev."""

    def __init__(self) -> None:
        self._store: dict[str, CachedResult] = {}
        self._lock = asyncio.Lock()

    async def get(self, key: str) -> CachedResult | None:
        async with self._lock:
            entry = self._store.get(key)
            if entry is None:
                return None
            if entry.expires_at <= time.monotonic():
                del self._store[key]
                return None
            return entry

    async def set(self, key: str, value: CachedResult, ttl: int) -> None:
        async with self._lock:
            value.expires_at = time.monotonic() + ttl
            self._store[key] = value

    async def delete(self, key: str) -> None:
        async with self._lock:
            self._store.pop(key, None)


async def get_verification(
    identifier: ProductIdentifier,
    cache: CacheBackend,
    call_vrs: Callable[[ProductIdentifier], Awaitable[dict]],
) -> dict:
    """Serve a VRS verification from cache when policy allows, else call through."""
    key = identifier.cache_key()
    cached = await cache.get(key)
    if cached is not None:
        return {**cached.payload, "cache_hit": True}

    result = await call_vrs(identifier)               # circuit-broken client call
    status = VerificationStatus(result["status"])
    ttl = TTL_SECONDS.get(status)                      # None for SUSPECT

    if ttl is not None:
        await cache.set(key, CachedResult(status, result, expires_at=0.0), ttl)
    # SUSPECT falls through uncached, every single time.

    return {**result, "cache_hit": False}

DSCSA/GS1 note: TTL_SECONDS.get(status) returning None for SUSPECT is the load-bearing line — it means a suspect result can never be written to the cache no matter what future status values get added carelessly to the dict, because the default is “do not cache” rather than “cache forever.”

Step 4 — Invalidate immediately on a status-change signal #

TTL expiry is a backstop, not the primary invalidation mechanism. The moment Suspect Product Investigation Workflows flags a serial, or a decommission event lands for a unit whose verification was cached as verified, the cached entry must be removed immediately rather than waiting out its TTL.

async def invalidate_on_status_change(
    identifier: ProductIdentifier, cache: CacheBackend
) -> None:
    """Called by the investigation workflow the instant a serial's status changes."""
    await cache.delete(identifier.cache_key())


async def handle_status_change_event(event: dict, cache: CacheBackend) -> None:
    identifier = ProductIdentifier(
        gtin=event["gtin"], serial=event["serial"],
        lot=event["lot"], expiry=event["expiry"],
    )
    await invalidate_on_status_change(identifier, cache)

DSCSA/GS1 note: wiring this handler to the same event stream that feeds suspect-product quarantine means the cache can never be the reason a dispenser is told a recalled or suspect unit is still verified — the cache is subordinate to the investigation workflow, never independent of it.

Step 5 — Fail open toward re-verification, never toward a stale positive #

If the cache backend itself is unreachable — a Redis outage, a network blip — the wrapper must behave as a cache miss, not raise and block verification, and it must never serve an entry past its TTL just because deleting it failed.

async def get_verification_safe(
    identifier: ProductIdentifier,
    cache: CacheBackend,
    call_vrs: Callable[[ProductIdentifier], Awaitable[dict]],
) -> dict:
    try:
        return await get_verification(identifier, cache, call_vrs)
    except (ConnectionError, asyncio.TimeoutError):
        # Cache backend is down: skip it entirely and call through.
        # Under-caching costs latency; over-caching costs correctness.
        result = await call_vrs(identifier)
        return {**result, "cache_hit": False, "cache_bypassed": True}

DSCSA/GS1 note: this asymmetry — degrade toward calling the real VRS endpoint, never toward trusting an unreachable cache — mirrors the same principle the circuit breaker guide applies to the partner call itself: a broken dependency should degrade capacity, not silently degrade correctness.

Verification #

Three checks confirm the cache is safe to run in production, not just fast:

import pytest

class FakeCache(InProcessTTLCache):
    pass

async def fake_vrs_verified(identifier: ProductIdentifier) -> dict:
    return {"status": "verified", "gtin": identifier.gtin, "serial": identifier.serial}

async def fake_vrs_suspect(identifier: ProductIdentifier) -> dict:
    return {"status": "suspect", "gtin": identifier.gtin, "serial": identifier.serial}


@pytest.mark.asyncio
async def test_verified_result_is_cached_and_reused():
    cache = FakeCache()
    ident = ProductIdentifier(gtin="00312345000010", serial="A1", lot="L1", expiry="271231")
    first = await get_verification(ident, cache, fake_vrs_verified)
    second = await get_verification(ident, cache, fake_vrs_verified)
    assert first["cache_hit"] is False
    assert second["cache_hit"] is True


@pytest.mark.asyncio
async def test_suspect_result_is_never_cached():
    cache = FakeCache()
    ident = ProductIdentifier(gtin="00312345000010", serial="A2", lot="L1", expiry="271231")
    await get_verification(ident, cache, fake_vrs_suspect)
    assert await cache.get(ident.cache_key()) is None


@pytest.mark.asyncio
async def test_invalidation_clears_a_verified_entry():
    cache = FakeCache()
    ident = ProductIdentifier(gtin="00312345000010", serial="A3", lot="L1", expiry="271231")
    await get_verification(ident, cache, fake_vrs_verified)
    await invalidate_on_status_change(ident, cache)
    assert await cache.get(ident.cache_key()) is None

Run these against the in-process backend in CI, then rerun the same suite against a real Redis instance — a CacheBackend implementation backed by redis.asyncio should pass identically, which is the point of hiding storage behind the protocol. In production, alert on any non-zero count of suspect-status cache writes; that metric should be structurally zero, and a single occurrence means Step 3’s TTL table was bypassed somewhere.

Gotchas & Edge Cases #

• Reordering the cache-key fields. Changing the concatenation order in cache_key() between deployments silently orphans every previously cached entry — harmless for correctness, but it causes a temporary cache-miss storm that looks like a partner outage. Version the key format explicitly (vrs:v2:verify:<hash>) if you ever need to change it.
• Treating “not found in cache” as “not verified.” A cache miss must always fall through to a real VRS call; never let an empty cache result be interpreted as a negative verification, or an idle cache becomes a false suspect-product signal.
• Caching across trading partners with the same GTIN pool. If more than one manufacturer’s router shares this cache layer, namespace the key by the resolving partner GLN as well, or a verification answered by partner A can be served for the same GTIN routed to partner B.
• Clock skew between router replicas. An in-process TTL cache uses time.monotonic() per process, which is correct locally but means two replicas can expire the same key at slightly different wall-clock moments. This is acceptable for TTL expiry but is exactly why invalidation (Step 4) — not TTL — is the mechanism that must fire on an actual status change.
• Leading zeros in GTIN or lot. Never coerce gtin or lot to an integer before hashing the cache key; a stripped leading zero produces a different key for the same physical unit and defeats both caching and invalidation.

Related #

• Up to the parent section: Verification Router Service Architecture
• Implementing circuit breakers for VRS API calls — the resilience layer this cache sits alongside on the same call path
• Suspect Product Investigation Workflows — the workflow whose status-change signal drives Step 4’s cache invalidation
• DSCSA Compliance Architecture & Standards Mapping — the section this guide belongs to