SerializationAutomation DSCSA Track & Trace

Enforcing Six-Year Retention with Python

A wholesale distributor’s EPCIS repository has grown to nine billion events across four years of packaging output, storage costs are climbing linearly, and nobody can say with confidence which records are actually eligible for deletion versus which ones sit under an open recall investigation. That is the exact problem this guide solves: implementing the DSCSA obligation to retain Transaction Information and Transaction Statements for six years, expressed as a scheduled Python job that tiers aging records to cheaper storage, blocks deletion under legal hold, and expires only what has genuinely passed its window — without ever breaking the hash-chained audit trail. It is a code-first deep dive within EPCIS Repository Storage & Retention, and it feeds directly into the broader Serialization Data Ingestion & EPCIS Event Sync pipeline that governs how events land in the repository in the first place.

Retention is not a single TTL. Different record classes carry different retention triggers under DSCSA, some events remain relevant to open investigations well past their nominal window, and a repository that is also the system of record for verification lookups cannot simply truncate old partitions the way a cache would. The sections below build a policy table, a tiering strategy, a legal-hold gate, and the sweeper itself.

Prerequisites

  • Python 3.10+ — snippets use X | Y union types and structural pattern matching.
  • asyncpg or an equivalent async database driver — the sweeper issues batched, cursor-paginated queries against the EPCIS repository rather than one giant DELETE.
  • aioboto3 or an S3-compatible async client — for archiving cold records to object storage before they are purged from the hot store.
  • Pydantic v2 — to model the retention policy and the legal-hold record as typed, validated structures rather than loose dictionaries.
  • DSCSA data prerequisites — every record must already carry record_class (e.g. transaction_information, object_event, aggregation_event), an event_time or transaction_date, a GTIN (01) and serial (21) pair for correlation, and the prior entry’s hash from the audit chain described in Building hash-chained audit logs for 21 CFR Part 11.
  • A legal-hold table or service populated by the suspect-product and investigation workflow, so the sweeper can query holds by GTIN/serial or by lot without reimplementing that logic.

Before writing the sweeper, fix the retention clock per record class. DSCSA measures the six-year window from the transaction date for Transaction Information and Transaction Statements, but ObjectEvent and AggregationEvent records are typically retained on the same six-year clock anchored to event_time so the full EPCIS lineage for a unit survives as long as its transaction record does. Getting this anchor wrong — for example, resetting the clock on every read or aggregation touch — silently extends retention forever and defeats the point of tiering.

Step-by-Step Solution

Step 1 — Model the retention policy per record class

Encode the retention window and storage tiers as data, not as scattered constants, so the policy can be audited and changed without touching sweeper logic.

from datetime import date, datetime, timedelta, timezone
from enum import Enum
from pydantic import BaseModel, field_validator


class RecordClass(str, Enum):
    TRANSACTION_INFORMATION = "transaction_information"
    OBJECT_EVENT = "object_event"
    AGGREGATION_EVENT = "aggregation_event"
    TRANSFORMATION_EVENT = "transformation_event"


class RetentionPolicy(BaseModel):
    record_class: RecordClass
    retention_years: int = 6          # DSCSA baseline for transaction records
    hot_days: int = 180                # served from the primary repository
    warm_days: int = 730               # served from a cheaper indexed tier
    # anything older than warm_days but inside retention_years moves to cold

    @field_validator("retention_years")
    @classmethod
    def _min_dscsa_window(cls, v: int) -> int:
        if v < 6:
            raise ValueError("DSCSA requires at least a six-year retention window")
        return v


DEFAULT_POLICIES: dict[RecordClass, RetentionPolicy] = {
    RecordClass.TRANSACTION_INFORMATION: RetentionPolicy(
        record_class=RecordClass.TRANSACTION_INFORMATION, hot_days=90, warm_days=365
    ),
    RecordClass.OBJECT_EVENT: RetentionPolicy(
        record_class=RecordClass.OBJECT_EVENT, hot_days=180, warm_days=730
    ),
    RecordClass.AGGREGATION_EVENT: RetentionPolicy(
        record_class=RecordClass.AGGREGATION_EVENT, hot_days=180, warm_days=730
    ),
    RecordClass.TRANSFORMATION_EVENT: RetentionPolicy(
        record_class=RecordClass.TRANSFORMATION_EVENT, hot_days=180, warm_days=730
    ),
}


def retention_cutoff(policy: RetentionPolicy, as_of: date | None = None) -> date:
    """Return the date before which records in this class are eligible for expiry."""
    as_of = as_of or datetime.now(timezone.utc).date()
    return as_of - timedelta(days=policy.retention_years * 365)

DSCSA/GS1 note: keeping retention_years as a validated floor rather than a raw config value means a misconfigured deployment cannot accidentally set a shorter-than-legal window — the validator raises before the policy ever loads.

A legal hold — opened by a suspect-product investigation, a recall, or a litigation notice — must categorically block deletion, independent of age. Query it as a first-class check, and fail closed: if the hold service is unreachable, treat the record as held rather than risk deleting something under investigation.

import logging

logger = logging.getLogger("retention.legal_hold")


class LegalHoldError(Exception):
    """Raised when the hold service cannot be reached; sweeper must fail closed."""


async def is_under_legal_hold(conn, gtin: str, serial: str, lot: str | None) -> bool:
    """Return True if any active hold matches this record's identifiers."""
    try:
        row = await conn.fetchrow(
            """
            SELECT 1 FROM legal_holds
            WHERE released_at IS NULL
              AND (
                    (gtin = $1 AND serial = $2)
                 OR (lot = $3 AND lot IS NOT NULL)
              )
            LIMIT 1
            """,
            gtin, serial, lot,
        )
        return row is not None
    except Exception as exc:  # connection error, timeout, etc.
        raise LegalHoldError(f"hold lookup failed for {gtin}/{serial}") from exc

DSCSA/GS1 note: matching on both the exact GTIN (01)/serial (21) pair and on lot (10) catches both unit-level holds from an active investigation and lot-level holds issued during a recall, so a single serial being investigated does not leave its sibling units in the same lot exposed to premature deletion.

Step 3 — Tier hot records to a cold object-storage archive

Before a record is ever eligible for deletion, it should move from the primary repository to progressively cheaper storage while remaining fully queryable by identifier. Archiving to object storage keeps the record retrievable for the remainder of the six-year window without paying primary-database costs for data nobody queries daily.

import json
import aioboto3


async def archive_to_cold_storage(
    s3_client, bucket: str, record: dict, record_class: RecordClass
) -> str:
    """Write one record to cold object storage, keyed for point lookups."""
    key = (
        f"{record_class.value}/"
        f"{record['gtin']}/{record['serial_number']}/"
        f"{record['event_time']}.json"
    )
    body = json.dumps(record, default=str).encode("utf-8")
    await s3_client.put_object(
        Bucket=bucket,
        Key=key,
        Body=body,
        ContentType="application/json",
        Metadata={"record-class": record_class.value, "audit-hash": record["hash"]},
    )
    return key


async def tier_aging_records(conn, s3_client, bucket: str, policy: RetentionPolicy) -> int:
    """Move records past warm_days into cold storage; return count archived."""
    cutoff = datetime.now(timezone.utc).date() - timedelta(days=policy.warm_days)
    rows = await conn.fetch(
        """
        SELECT * FROM epcis_events
        WHERE record_class = $1 AND event_time < $2 AND storage_tier = 'warm'
        ORDER BY event_time
        LIMIT 5000
        """,
        policy.record_class.value, cutoff,
    )
    archived = 0
    async with aioboto3.Session().client("s3") as _:  # pooled per-call in production
        for row in rows:
            record = dict(row)
            key = await archive_to_cold_storage(s3_client, bucket, record, policy.record_class)
            await conn.execute(
                "UPDATE epcis_events SET storage_tier = 'cold', cold_key = $1 WHERE id = $2",
                key, record["id"],
            )
            archived += 1
    return archived

DSCSA/GS1 note: the archive preserves the original record verbatim, including its position in the hash chain (audit-hash metadata), so a record moved to cold storage remains reproducible in its exact EPCIS structure if a trading partner or inspector requests it later.

The sweeper is the component that actually deletes. It must page through candidates rather than issue a single unbounded query, skip anything under legal hold, and default to a dry run so a policy misconfiguration is caught in a report before it touches data.

import asyncio
from dataclasses import dataclass, field


@dataclass
class SweepResult:
    record_class: RecordClass
    scanned: int = 0
    eligible: int = 0
    held: int = 0
    deleted: int = 0
    errors: list[str] = field(default_factory=list)


async def sweep_expired_records(
    conn,
    policy: RetentionPolicy,
    *,
    batch_size: int = 500,
    dry_run: bool = True,
) -> SweepResult:
    """Expire records past the DSCSA retention window, honoring legal holds."""
    cutoff = retention_cutoff(policy)
    result = SweepResult(record_class=policy.record_class)

    while True:
        rows = await conn.fetch(
            """
            SELECT id, gtin, serial_number, lot_number
            FROM epcis_events
            WHERE record_class = $1 AND event_time < $2 AND expired_at IS NULL
            ORDER BY id
            LIMIT $3
            """,
            policy.record_class.value, cutoff, batch_size,
        )
        if not rows:
            break

        result.scanned += len(rows)
        for row in rows:
            try:
                held = await is_under_legal_hold(
                    conn, row["gtin"], row["serial_number"], row["lot_number"]
                )
            except LegalHoldError as exc:
                result.errors.append(str(exc))
                continue  # fail closed: skip, never delete on uncertainty

            if held:
                result.held += 1
                continue

            result.eligible += 1
            if not dry_run:
                # Mark expired rather than hard-delete: keeps the audit chain's
                # prior-hash links intact for any record that still references it.
                await conn.execute(
                    "UPDATE epcis_events SET expired_at = now() WHERE id = $1",
                    row["id"],
                )
                result.deleted += 1

        await asyncio.sleep(0)  # yield control between batches

    return result


async def run_retention_sweep(conn, policies: dict[RecordClass, RetentionPolicy], dry_run: bool = True):
    """Entry point for a scheduled job — one sweep per record class."""
    results = []
    for policy in policies.values():
        outcome = await sweep_expired_records(conn, policy, dry_run=dry_run)
        results.append(outcome)
        logger.info(
            "sweep class=%s scanned=%d eligible=%d held=%d deleted=%d dry_run=%s",
            outcome.record_class.value, outcome.scanned, outcome.eligible,
            outcome.held, outcome.deleted, dry_run,
        )
    return results

DSCSA/GS1 note: expiring by setting expired_at instead of issuing a DELETE preserves every hash-chain link for records that were legitimately retained before their own expiry, so a downstream integrity check against hash-chained audit logs never encounters a dangling prev_hash reference. Only a separate, slower vacuum process — run well after expired_at is set and double-checked against legal holds a second time — should physically remove the row from disk.

Step 5 — Schedule the sweep and default it to dry-run

Wire the sweep into a scheduler that runs nightly, and require an explicit flag to leave dry-run mode, so a first deployment always surfaces its intended impact before it deletes anything.

import argparse


async def main() -> None:
    parser = argparse.ArgumentParser(description="DSCSA retention sweeper")
    parser.add_argument("--live", action="store_true", help="actually expire records")
    args = parser.parse_args()

    conn = await get_repository_connection()  # project-specific pool acquisition
    try:
        results = await run_retention_sweep(conn, DEFAULT_POLICIES, dry_run=not args.live)
        total_eligible = sum(r.eligible for r in results)
        total_held = sum(r.held for r in results)
        if not args.live and total_eligible:
            logger.warning(
                "dry run found %d expirable records (%d under legal hold, skipped) — "
                "rerun with --live to apply",
                total_eligible, total_held,
            )
    finally:
        await conn.close()


if __name__ == "__main__":
    asyncio.run(main())

DSCSA/GS1 note: defaulting to dry_run=True and requiring --live mirrors the same fail-safe posture as retrying only transient failures elsewhere in the pipeline — a scheduling mistake should produce a loud report, never a silent bulk deletion of records still inside the DSCSA window.

Verification

Confirm the sweeper’s three guarantees with targeted tests: it never expires anything inside the window, it never deletes anything under hold, and dry-run mode never mutates state.

import pytest


@pytest.mark.asyncio
async def test_dry_run_makes_no_changes(fake_conn_with_expired_records):
    result = await sweep_expired_records(
        fake_conn_with_expired_records,
        DEFAULT_POLICIES[RecordClass.OBJECT_EVENT],
        dry_run=True,
    )
    assert result.eligible > 0
    assert result.deleted == 0  # dry run must not touch expired_at


@pytest.mark.asyncio
async def test_legal_hold_blocks_expiry(fake_conn_with_one_held_record):
    result = await sweep_expired_records(
        fake_conn_with_one_held_record,
        DEFAULT_POLICIES[RecordClass.TRANSACTION_INFORMATION],
        dry_run=False,
    )
    assert result.held == 1
    assert result.deleted == 0


def test_retention_cutoff_is_six_years_back():
    policy = DEFAULT_POLICIES[RecordClass.TRANSACTION_INFORMATION]
    cutoff = retention_cutoff(policy, as_of=date(2026, 7, 17))
    assert cutoff.year == 2020

Beyond unit tests, run the sweeper against a staging snapshot with --live omitted and diff its report against a manual SELECT count(*) for the same cutoff and hold conditions — the counts must match exactly. Before enabling --live in production, confirm the audit-log integrity check still passes end to end so that setting expired_at on old rows has not broken any prev_hash link a still-active record depends on.

Gotchas & Edge Cases

  • Anchoring the clock to the wrong timestamp. Resetting event_time on aggregation or re-indexing operations silently restarts the six-year clock and means records never expire. Anchor strictly to the original commissioning or transaction timestamp, in UTC, and never overwrite it.
  • Hold-service unavailability treated as “no hold.” If the legal-hold lookup times out, the sweeper must skip the record, not assume it is clear — a fail-open design is the single most dangerous mistake in this workflow.
  • Hard-deleting instead of marking expired. A physical DELETE on a record that another entry’s hash chain still references breaks verifiability for every record after it. Always mark expired_at first and let a separate, carefully audited vacuum step do physical removal.
  • Idempotency-key collisions on rerun. If the sweep is interrupted mid-batch and rerun, records already marked expired_at must be excluded from the next scan (the expired_at IS NULL filter above), or the job will attempt to re-expire already-expired rows and inflate its own metrics.
  • Cold-archive keys that don’t match the hot-store identifier scheme. If the S3 key layout doesn’t mirror the (gtin, serial, event_time) lookup path used by indexing EPCIS events for sub-second verification, a verification request for an archived unit has no deterministic way to find its cold copy.

Continue reading

Related Indexing for Sub-Second Lookups