Automating Decommission Events in Python

A vision-system reject, a QC hold, or a stability-sampling pull all end the same way: a serialized unit must be permanently retired from the supply chain with an auditable EPCIS record, not just a status flag flipped in an ERP table. This guide is a code-first companion to Decommission & Reaggregation Rules within Aggregation Hierarchy & Validation Workflows, and it narrows in on the two failure modes that cause the most damage in production: a decommission event generated for a serial that was never legitimately commissioned, and a decommission that orphans children still bound to the unit under aggregation. Both defects look identical to the packaging line — the physical unit leaves the process — but one of them silently breaks the traceability chain a trading partner or an FDA inspector will later query.

The fix is a small, disciplined contract: a Pydantic v2 model that only ever produces a well-formed ObjectEvent with action: DELETE, a guard function that refuses to build that event unless the serial has a prior commission record, and a cascade routine that walks the aggregation hierarchy so every child is decommissioned before its container. The diagram below is the shape of that contract before any code runs.

Prerequisites #

• Python 3.10+ — the code below uses X | None union syntax and list[str] generics.
• Pydantic v2 (pydantic>=2.0) — field_validator enforces the EPCIS syntax rules structurally rather than through scattered if checks.
• An async repository client — anything exposing get_commission_event_time, get_serial_state, and get_aggregated_children against your EPCIS store; the examples treat it as an injected dependency so the guard and cascade logic stay storage-agnostic.
• DSCSA data prerequisites — a commission history you can query per SGTIN (GTIN (01) + serial (21)), and an up-to-date parent-child aggregation graph, maintained by the same layer documented in Parent-Child Serial Mapping.
• A dead-letter sink — a queue or table that captures rejected decommission requests with their reason, so a guard failure is reported, not swallowed.

Step-by-Step Solution #

Step 1 — Model the decommission event contract #

Define the reasons a unit gets decommissioned and a Pydantic v2 model that can only ever serialize into a structurally valid ObjectEvent. The model pins action to DELETE, the business step to decommissioning, and the disposition to urn:epcglobal:cbv:disp:inactive for the damaged/destroyed/sampled paths this guide covers — physical-destruction-with-certificate cases that warrant urn:epcglobal:cbv:disp:destroyed instead are a disposition-selection decision made upstream, not something this builder second-guesses.

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


class DecommissionReason(str, Enum):
    DAMAGED = "damaged"
    DESTROYED = "destroyed"
    SAMPLED = "sampled"


class DecommissionEvent(BaseModel):
    epc: str                     # SGTIN URI built from AI (01) + (21)
    action: str = "DELETE"
    biz_step: str = "urn:epcglobal:cbv:bizstep:decommissioning"
    disposition: str = "urn:epcglobal:cbv:disp:inactive"
    event_time: datetime
    reason: DecommissionReason
    parent_epc: str | None = None  # set only when cascaded from a container

    @field_validator("action")
    @classmethod
    def _action_is_delete(cls, v: str) -> str:
        if v != "DELETE":
            raise ValueError("a decommission ObjectEvent must carry action DELETE")
        return v

    @field_validator("event_time")
    @classmethod
    def _tz_aware(cls, v: datetime) -> datetime:
        # EPCIS requires an explicit offset; a naive timestamp is unauditable.
        if v.tzinfo is None:
            raise ValueError("event_time must carry a timezone offset")
        return v.astimezone(timezone.utc)

    def to_epcis_payload(self) -> dict:
        payload = {
            "type": "ObjectEvent",
            "action": self.action,
            "bizStep": self.biz_step,
            "disposition": self.disposition,
            "epcList": [self.epc],
            "eventTime": self.event_time.isoformat(),
            "extensions": {"reasonCode": self.reason.value},
        }
        if self.parent_epc:
            payload["extensions"]["cascadedFrom"] = self.parent_epc
        return payload

DSCSA/GS1 note: pinning action and bizStep as validated fields rather than free strings means a typo or a copy-pasted ADD from a commission template fails at model construction, before it ever reaches the L4 repository.

Step 2 — Guard: refuse a decommission that isn’t sequenced correctly #

The single most important invariant in this workflow is that a decommission ObjectEvent can never be committed before the commission event it supersedes, and a serial already in a terminal state can never be decommissioned twice. Both checks belong in one guard so every call site enforces them the same way.

class DecommissionSequenceError(Exception):
    """Raised when a decommission would precede or duplicate a commission."""


async def assert_commission_precedes(
    repo, epc: str, decommission_time: datetime
) -> datetime:
    commission_time = await repo.get_commission_event_time(epc)
    if commission_time is None:
        raise DecommissionSequenceError(
            f"{epc} has no commission record; refusing to decommission"
        )
    if decommission_time <= commission_time:
        raise DecommissionSequenceError(
            f"{epc} decommission at {decommission_time.isoformat()} cannot "
            f"precede its commission at {commission_time.isoformat()}"
        )
    state = await repo.get_serial_state(epc)
    if state == "DECOMMISSIONED":
        raise DecommissionSequenceError(f"{epc} is already DECOMMISSIONED")
    return commission_time

DSCSA/GS1 note: this is the same ordering guarantee the ingestion side gets from partitioning a Kafka topic by (gtin, serial) — here it is enforced explicitly, at the point of event generation, so it holds even if the underlying broker ever delivers out of order.

Step 3 — Cascade to children before the parent #

A container-level decommission — a whole case pulled for quality-control sampling — must not leave its children logically bound to a parent that the record now claims is gone. The cascade walks the aggregation hierarchy owned by Parent-Child Serial Mapping and decommissions every child before the container itself, so the audit trail always shows the leaf units retiring first.

async def cascade_decommission(
    repo,
    root_epc: str,
    reason: DecommissionReason,
    event_time: datetime,
    _parent: str | None = None,
) -> list[DecommissionEvent]:
    """Post-order walk: every child is decommissioned before its parent."""
    events: list[DecommissionEvent] = []
    for child_epc in await repo.get_aggregated_children(root_epc):
        events.extend(
            await cascade_decommission(repo, child_epc, reason, event_time, root_epc)
        )
    await assert_commission_precedes(repo, root_epc, event_time)
    events.append(
        DecommissionEvent(
            epc=root_epc,
            event_time=event_time,
            reason=reason,
            parent_epc=_parent,
        )
    )
    return events

DSCSA/GS1 note: running the commission guard at every recursion level — not just on the root — satisfies GS1 EPCIS aggregation integrity for the whole subtree; a container built from a serial that itself lacks a commission record must fail the same way the root would. Containers assembled under Case & Pallet Aggregation Logic are exactly the shape this walk expects to traverse.

Step 4 — Emit idempotently and preserve the audit trail #

Push the cascade’s events in order, deriving the idempotency key from the serial and reason so a retried request returns the original response instead of duplicating the ObjectEvent. Each successful push also appends a hash-linked audit-log entry, matching the retention discipline the parent guide requires.

import hashlib


async def emit_cascade(db, l4_repository, events: list[DecommissionEvent]) -> list[dict]:
    """Push every decommission event in cascade order, skipping already-recorded ones."""
    acks: list[dict] = []
    async with db.transaction():
        for evt in events:
            idempotency_key = hashlib.sha256(
                f"{evt.epc}:{evt.reason.value}".encode()
            ).hexdigest()[:24]

            existing = await db.fetch_event(idempotency_key)
            if existing:
                acks.append(existing)
                continue

            await db.update_serial_state(evt.epc, "DECOMMISSIONED")
            response = await l4_repository.push(evt.to_epcis_payload())
            await db.store_event_log(idempotency_key, response)
            acks.append(response)
    return acks

DSCSA/GS1 note: keeping the idempotency key free of the timestamp is what makes a network retry safe — a retried push computes the same key and short-circuits to the stored response rather than emitting a second decommissioning record for the same physical action.

Step 5 — Wire it into a non-blocking request handler #

The line-facing entry point ties the guard, cascade, and emission together, and routes any sequencing failure to the dead-letter sink instead of raising into the packaging process. A rejected request is a compliance signal, not a crash.

async def handle_decommission_request(
    db, l4_repository, epc: str, reason: DecommissionReason, dlq
) -> None:
    event_time = datetime.now(timezone.utc)
    try:
        events = await cascade_decommission(db, epc, reason, event_time)
    except DecommissionSequenceError as exc:
        await dlq.put({"epc": epc, "reason": reason.value, "error": str(exc)})
        return
    await emit_cascade(db, l4_repository, events)

DSCSA/GS1 note: separating rejection from failure this way mirrors the exception-handling posture the broader ingestion pipeline uses — a sequencing violation is escalated for review, while a transient repository timeout should be retried with backoff rather than routed here at all.

Verification #

Confirm the guard and cascade are both correct before pointing this at a live line. Three checks give fast, high-confidence signal:

import pytest
from datetime import datetime, timedelta, timezone


class FakeRepo:
    def __init__(self, commission_times, states, children):
        self.commission_times = commission_times
        self.states = states
        self.children = children

    async def get_commission_event_time(self, epc):
        return self.commission_times.get(epc)

    async def get_serial_state(self, epc):
        return self.states.get(epc, "AGGREGATED")

    async def get_aggregated_children(self, epc):
        return self.children.get(epc, [])


@pytest.mark.asyncio
async def test_guard_rejects_uncommissioned_serial():
    repo = FakeRepo(commission_times={}, states={}, children={})
    with pytest.raises(DecommissionSequenceError):
        await assert_commission_precedes(repo, "epc:never-commissioned", datetime.now(timezone.utc))


@pytest.mark.asyncio
async def test_guard_rejects_decommission_before_commission():
    now = datetime.now(timezone.utc)
    repo = FakeRepo(commission_times={"epc:1": now}, states={}, children={})
    with pytest.raises(DecommissionSequenceError):
        await assert_commission_precedes(repo, "epc:1", now - timedelta(seconds=1))


@pytest.mark.asyncio
async def test_cascade_orders_children_before_parent():
    now = datetime.now(timezone.utc)
    commission_times = {"case:1": now - timedelta(days=1), "unit:1": now - timedelta(days=1)}
    repo = FakeRepo(commission_times=commission_times, states={}, children={"case:1": ["unit:1"]})
    events = await cascade_decommission(repo, "case:1", DecommissionReason.DAMAGED, now)
    assert [e.epc for e in events] == ["unit:1", "case:1"]

For a production audit check rather than a unit test, pull a sample of committed decommission records and reconcile each one’s eventTime against the commission eventTime for the same SGTIN — every row should show a strictly later decommission, and every container-level decommission should have at least one preceding child row in the same transaction. Any exception surfaced from the dead-letter sink should also be queryable by reason code, so a spike in DecommissionSequenceError rejections shows up as an operational metric, not just log noise.

Gotchas & Edge Cases #

• Leading-zero GTINs. The epc field is an SGTIN URI, not a bare integer — never cast the company-prefix or item-reference segments to int anywhere upstream of this model, or a stripped leading zero will corrupt the identifier the decommission event claims to retire.
• UTC vs. local event_time. The guard compares decommission_time against a stored commission_time; both must be timezone-aware and normalized to UTC before comparison, or a local-time commission and a UTC decommission can produce a false-positive sequencing error (or worse, a false negative that lets an out-of-order event through).
• Idempotency-key collisions across incidents. Deriving the key from epc plus reason is safe for retries of the same incident, but if a unit is legitimately re-flagged under the same reason code after a prior rejection, the key collides with the earlier attempt. Include an incident or batch identifier in the key when your process allows repeat submissions for one serial.
• Cyclic or unbounded aggregation graphs. cascade_decommission recurses on get_aggregated_children; a corrupted hierarchy with a cycle will recurse indefinitely. Guard the repository query layer with a maximum depth or a visited-set check before this ever reaches production data.
• Repackaged units re-entering the graph. A serial decommissioned here should never resurface as a child in a later aggregation; if your process instead repackages surviving stock into new containers, that is a distinct workflow — see Handling Reaggregation After Repackaging for how that boundary is enforced.

Related #

• Up to the parent topic: Decommission & Reaggregation Rules
• Handling Reaggregation After Repackaging — the sibling workflow for surviving stock that re-enters a new hierarchy
• Parent-Child Serial Mapping — the aggregation graph this cascade reads and mutates
• Aggregation Hierarchy & Validation Workflows — the domain covering hierarchy nesting and validation this guide operates within