Automating Case-to-Pallet Aggregation Validation in Python

When a packaging line closes a pallet, it commits a promise: that the single Serial Shipping Container Code (SSCC) printed on the pallet label truthfully expands into every case SSCC beneath it, and every case into its unit SGTINs. A downstream wholesaler will scan that one pallet code and infer the thousand units inside it — never rescanning them — so any miscount, duplicate, or orphaned serial silently poisons the transaction data the Drug Supply Chain Security Act (DSCSA) requires trading partners to exchange. This page is the concrete Python implementation of the workflow described in Case & Pallet Aggregation Logic: a validation gate that intercepts each aggregation payload from the line, proves the parent-child tree is complete and unique before it is allowed to exist, and emits an immutable AggregationEvent that a partner can trust. Get the gate right and outages degrade gracefully; get it wrong and product is quarantined at the receiving dock while verification requests return INVALID.

Prerequisites #

Before wiring the validator into a line, confirm you have the following in place:

• Python 3.10+ — the code below uses match statements, list[str] builtin generics, and datetime timezone handling.
• Pydantic v2 (pip install "pydantic>=2") — for typed schema enforcement with field_validator. Identifier rules follow the contracts described in GS1 Standards Implementation.
• redis-py (pip install "redis>=5") — for the distributed state store that tracks which serial belongs to which parent across concurrent line workers. An in-memory dict works for a single process, but any real line runs multiple consumers.
• A message consumer — Kafka, RabbitMQ, or MQTT delivering raw aggregation payloads from the L3 line controller. The broker-backed ingestion pattern is covered under Async Batch Processing Pipelines.
• DSCSA data prerequisites: an active, commissioned SGTIN pool (units already validated per the (01)/(21)/(17)/(10) DataMatrix contract), a managed SSCC pool carrying AI (00) for cases and pallets, the trading-partner GLN for the packaging site, and the expected pack configuration (units-per-case, cases-per-pallet) from the master data record.

Step-by-step solution #

Step 1 — Model the aggregation payload with typed validation #

Nothing enters the hierarchy until it survives schema enforcement. A single Pydantic model normalizes the payload, strips whitespace, and rejects a malformed SSCC or SGTIN URN on contact.

from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator


class AggregationPayload(BaseModel):
    parent_sscc: str = Field(..., description="18-digit SSCC, GS1 AI (00)")
    child_epcs: list[str] = Field(..., min_length=1)
    event_time: str
    gln: str
    line_id: str

    @field_validator("parent_sscc")
    @classmethod
    def validate_sscc(cls, v: str) -> str:
        digits = v.strip()
        if not digits.isdigit() or len(digits) != 18:
            raise ValueError("SSCC must be exactly 18 numeric digits")
        if not _gs1_check_digit_ok(digits):
            raise ValueError("SSCC check digit failed")
        return digits

    @field_validator("child_epcs", mode="before")
    @classmethod
    def validate_children(cls, v: list[str]) -> list[str]:
        cleaned = [c.strip() for c in v]
        for epc in cleaned:
            if not (epc.startswith("urn:epc:id:sgtin:")
                    or epc.startswith("urn:epc:id:sscc:")):
                raise ValueError(f"Unrecognized child EPC: {epc!r}")
        return cleaned

    @field_validator("event_time")
    @classmethod
    def validate_iso8601(cls, v: str) -> str:
        datetime.fromisoformat(v.replace("Z", "+00:00"))  # raises on malformed
        return v

Rule satisfied: the SSCC check digit and 18-digit length enforce GS1 General Specifications for AI (00); requiring child EPCs as GS1 URNs keeps the identifiers resolvable in EPCIS. The check-digit helper _gs1_check_digit_ok implements the standard modulo-10 algorithm — the same routine you validate GTINs with in mapping GTINs to NDCs.

Step 2 — Enforce cardinality against the pack configuration #

A case that should hold 50 units and reports 49 or 51 is not a smaller shipment — it is a defect. Validate the child count against master data before the container is allowed to close.

class PackConfig(BaseModel):
    units_per_case: int
    cases_per_pallet: int


def check_cardinality(payload: AggregationPayload, cfg: PackConfig) -> None:
    expected = cfg.cases_per_pallet if payload.parent_sscc_is_pallet else cfg.units_per_case
    actual = len(payload.child_epcs)
    if actual != expected:
        raise AggregationError("ERR_CARDINALITY_MISMATCH",
                               f"expected {expected} children, got {actual}")

Rule satisfied: DSCSA saleable-unit traceability depends on complete containment. A cardinality mismatch means unaccounted product, so the container must never emit an event until the count is exact — a full-count gate rather than an after-the-fact reconciliation. Tuning how hard this gate stops the line is the subject of Threshold Tuning for Line Speeds.

Step 3 — Guarantee uniqueness across the active batch #

No serial may belong to two parents at once. This is where a single-process set is not enough: concurrent workers must consult a shared store. Redis gives O(1) membership tests and an atomic claim.

import redis.asyncio as aioredis


async def claim_children(r: aioredis.Redis, payload: AggregationPayload) -> None:
    # Atomically claim every child for this parent; fail if any is already owned.
    async with r.pipeline(transaction=True) as pipe:
        for epc in payload.child_epcs:
            pipe.set(f"owner:{epc}", payload.parent_sscc, nx=True)
        results = await pipe.execute()

    duplicates = [epc for epc, ok in zip(payload.child_epcs, results) if not ok]
    if duplicates:
        # Roll back the ones we did claim, then reject.
        await r.delete(*[f"owner:{e}" for e, ok in zip(payload.child_epcs, results) if ok])
        raise AggregationError("ERR_DUPLICATE_CHILD",
                               f"already aggregated: {duplicates}")

Rule satisfied: uniqueness is what makes partner inference trustworthy — a duplicated child corrupts counts everywhere the pallet is scanned. The nx=True (set-if-not-exists) claim makes the check atomic across workers, so two lines can never both bind the same unit. Bidirectional integrity of these links is maintained by the Parent-Child Serial Mapping layer.

Step 4 — Validate temporal order and location context #

A pallet cannot be built before its cases are packed, and a case cannot be packed before its units are commissioned. Reject aggregation timestamps that precede the children’s own event times, and reject a (414) GLN that disagrees with the site.

def check_context(payload: AggregationPayload,
                  child_times: dict[str, datetime],
                  site_gln: str) -> None:
    agg_time = datetime.fromisoformat(payload.event_time.replace("Z", "+00:00"))
    for epc, t in child_times.items():
        if agg_time < t:
            raise AggregationError("ERR_TEMPORAL_DRIFT",
                                   f"{epc} serialized after aggregation")
    if payload.gln != site_gln:
        raise AggregationError("ERR_GLN_MISMATCH",
                               f"payload GLN {payload.gln} != site {site_gln}")

Rule satisfied: EPCIS event ordering must be chronologically defensible for an inspector to reconstruct product movement; a GLN mismatch signals a mis-routed payload from the wrong line. Both checks keep the emitted event auditable under DSCSA.

Step 5 — Emit an immutable, idempotent AggregationEvent #

Only after all four gates pass does an event get to exist. Build it in the EPCIS 2.0 shape and attach a deterministic idempotency key so a network retry never writes the record twice.

import hashlib


def build_event(payload: AggregationPayload) -> dict:
    key = hashlib.sha256(
        f"{payload.parent_sscc}:{','.join(sorted(payload.child_epcs))}".encode()
    ).hexdigest()[:24]

    return {
        "type": "AggregationEvent",
        "eventTime": payload.event_time,
        "eventTimeZoneOffset": "+00:00",
        "action": "ADD",
        "bizStep": "urn:epcglobal:cbv:bizstep:packing",
        "disposition": "urn:epcglobal:cbv:disp:in_progress",
        "parentID": f"urn:epc:id:sscc:{payload.parent_sscc}",
        "childEPCs": payload.child_epcs,
        "readPoint": {"id": f"urn:epc:id:sgln:{payload.gln}"},
        "idempotencyKey": key,
    }

Rule satisfied: the action of ADD, the packing business step, and a UTC eventTime produce a partner-consumable EPCIS 2.0 record; deriving the idempotency key from the parent plus the sorted child set (and deliberately excluding the timestamp) means a retried transmission reproduces the same key and de-duplicates at ingestion. The exact field-level serialization is detailed in the step-by-step guide to EPCIS 2.0 event formatting. A failed gate routes the payload instead to a dead-letter queue with a structured error code (ERR_DUPLICATE_CHILD, ERR_CARDINALITY_MISMATCH, ERR_TEMPORAL_DRIFT, ERR_GLN_MISMATCH), the quarantine pattern shared with Schema Validation & Error Handling.

Verification #

Prove the gate behaves before it ever touches a live line. A pytest scaffold exercises the happy path and each rejection:

import pytest

CFG = PackConfig(units_per_case=50, cases_per_pallet=40)


def make_payload(children: list[str], **kw):
    base = dict(parent_sscc="003123450000000018",
                child_epcs=children, event_time="2026-06-05T10:00:00Z",
                gln="0312345000009", line_id="L1")
    base.update(kw)
    return AggregationPayload(**base)


def test_cardinality_rejects_short_case():
    p = make_payload([f"urn:epc:id:sgtin:0312345.0.{i}" for i in range(49)])
    with pytest.raises(AggregationError) as e:
        check_cardinality(p, CFG)  # p.parent_sscc_is_pallet == False
    assert e.value.code == "ERR_CARDINALITY_MISMATCH"


def test_idempotency_key_is_order_independent():
    a = build_event(make_payload(["urn:epc:id:sgtin:1", "urn:epc:id:sgtin:2"]))
    b = build_event(make_payload(["urn:epc:id:sgtin:2", "urn:epc:id:sgtin:1"]))
    assert a["idempotencyKey"] == b["idempotencyKey"]

For end-to-end confidence, validate a batch of emitted events against the published EPCIS 2.0 JSON schema, then inspect the audit log: every payload the line produced must resolve to exactly one of two terminal outcomes — an emitted event with a unique idempotency key, or a dead-letter record with an error code. There is no third state. If the counts of (payloads in) and (events + dead-letters out) ever disagree, a gate is leaking.

Gotchas & edge cases #

• Leading zeros are data, not decoration. A 14-digit GTIN or an 18-digit SSCC with a stripped leading zero is a different identifier. Keep them as strings end to end; never let JSON or a spreadsheet coerce them to integers.
• UTC vs. local time in EPCIS. eventTime must be UTC while eventTimeZoneOffset records the local offset separately. Storing a local timestamp as if it were UTC is the most common cause of ERR_TEMPORAL_DRIFT false positives across a shift change or daylight-saving boundary.
• Idempotency keys that include the timestamp. A retry produces a new timestamp; if that timestamp feeds the key, the duplicate check is bypassed and you emit two events for one physical pallet. Derive the key only from stable identity — parent SSCC plus the sorted child set.
• Reusing an SSCC too soon. GS1 requires an SSCC not be reissued for at least a year. A managed pool that recycles case SSCCs within a batch will collide with a still-active pallet and trigger phantom ERR_DUPLICATE_CHILD rejections.
• Partial pallets at end of run. A legitimately short final pallet still fails a strict cardinality gate. Handle it as an explicit, operator-authorized exception with its own business step — never by loosening the count check, which would blind the gate to real defects. Breaking and rebuilding a hierarchy safely is governed by Decommission & Reaggregation Rules.

FAQ #

Why validate in Python middleware instead of on the L3 line controller? The controller is optimized for machine control, not for cross-batch state, historical uniqueness, or EPCIS shaping. A Python gate sits between L3 and the L4 repository as stateful validation middleware: it enforces the rules the repository cannot retroactively repair without breaking audit integrity, while leaving hardware timing untouched.

Does the validation have to be synchronous with the line? Tree building and the four gates must be synchronous and local so the line never blocks on the network, but transmission of the emitted event is asynchronous. Keeping the emit-or-quarantine decision in memory within a sub-100 ms budget, then buffering downstream, lets a network outage degrade gracefully instead of stopping packaging.

What happens to a payload that fails a gate? It routes to a dead-letter queue tagged with a machine-readable error code and never reaches the repository. Compliance officers get immediate root-cause data, and the quarantined hierarchy can be corrected and replayed without contaminating the clean event stream.

How do I break an aggregation later without violating immutability? You never edit or delete the original AggregationEvent. To disaggregate, append a compensating AggregationEvent with action of DELETE naming the same parent and the removed children, preserving a tamper-evident history for DSCSA.

Related #

• Case & Pallet Aggregation Logic — the parent workflow this implementation realizes.
• Parent-Child Serial Mapping — maintaining bidirectional link integrity and detecting drift.
• Step-by-Step Guide to EPCIS 2.0 Event Formatting — the exact serialization of the event this gate emits.
• Aggregation Hierarchy & Validation Workflows — the broader domain of nesting, validation gates, and EPCIS event generation.