Writing a Pydantic EPCIS 2.0 Validator from Scratch

A trading partner’s onboarding test batch arrives as EPCIS 2.0 JSON-LD, and your ingestion service accepts it with a shrug — no eventTimeZoneOffset, a bizStep that is not a CBV URI, an epcList entry that is not a well-formed SGTIN URI — and three weeks later a verification query returns nothing for a serial that should exist. This guide builds the fix: a Pydantic v2 model set that treats every inbound ObjectEvent as an untrusted payload until it passes typed structural checks, as part of Schema Validation & Error Handling, itself one piece of the wider Serialization Data Ingestion & EPCIS Event Sync pipeline. Unlike the XML-side parsing covered in parsing EPCIS XML with lxml, this is the JSON-LD ingestion boundary, and the goal is identical: a malformed event must fail loudly and land in a dead-letter queue, never half-persist as a legally insufficient record.

Prerequisites #

• Python 3.10+ — the models use X | None unions, Literal types, and Annotated field metadata.
• Pydantic v2 (pydantic>=2.0) — this guide relies on field_validator, model_validator, TypeAdapter, and discriminated unions via Field(discriminator=...), none of which exist in the v1 API.
• A JSON-LD deserializer — json.loads is sufficient for this guide; if events arrive over HTTP, httpx or aiohttp is the usual transport, feeding raw dicts into the models below.
• DSCSA data prerequisites — sample or production ObjectEvent payloads containing an epcList of SGTIN URIs (GTIN (01) + serial (21)), a bizStep, disposition, readPoint, and — for full pedigree — an ilmd block carrying lot (10) and expiry (17). The exact JSON-LD shape these validators expect is produced by the step-by-step guide to EPCIS 2.0 event formatting.
• A dead-letter sink — a queue, table, or topic that accepts {payload, errors, received_at} records so a ValidationError never simply vanishes into a log line.

Step-by-Step Solution #

Step 1 — Model the shared EPCIS envelope #

Every EPCIS 2.0 event — regardless of type — carries a timestamp pair, a business step, a disposition, and a read point. Put those on a base class so the four event models inherit identical timezone and CBV-URI enforcement instead of re-implementing it four times.

from __future__ import annotations

import re
from datetime import datetime
from typing import Literal

from pydantic import BaseModel, Field, field_validator, model_validator

CBV_URI_RE = re.compile(r"^urn:epcglobal:cbv:(bizstep|disp):[a-z0-9_]+$")
OFFSET_RE = re.compile(r"^[+-]\d{2}:\d{2}$")


class ReadPoint(BaseModel):
    id: str  # SGLN URI, e.g. urn:epc:id:sgln:0312345.00000.0


class EPCISEventBase(BaseModel):
    eventTime: datetime
    eventTimeZoneOffset: str
    bizStep: str
    disposition: str
    readPoint: ReadPoint

    @field_validator("eventTime")
    @classmethod
    def _tz_aware(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError(
                "eventTime must be a timezone-aware ISO 8601 timestamp"
            )
        return v

    @field_validator("eventTimeZoneOffset")
    @classmethod
    def _offset_syntax(cls, v: str) -> str:
        if not OFFSET_RE.fullmatch(v):
            raise ValueError("eventTimeZoneOffset must match ±HH:MM")
        return v

    @field_validator("bizStep", "disposition")
    @classmethod
    def _cbv_uri(cls, v: str) -> str:
        if not CBV_URI_RE.match(v):
            raise ValueError(f"expected a CBV bizstep/disp URI, got {v!r}")
        return v

    @model_validator(mode="after")
    def _offset_matches_event_time(self) -> "EPCISEventBase":
        # eventTime is UTC-normalized by Pydantic's datetime parser; the
        # declared offset must still match the timestamp's own offset so
        # a caller cannot claim "+05:30" while sending a "Z" timestamp.
        declared = self.eventTimeZoneOffset
        actual = self.eventTime.strftime("%z")
        actual_hhmm = f"{actual[:3]}:{actual[3:]}" if actual else ""
        if actual_hhmm and declared != actual_hhmm:
            raise ValueError(
                f"eventTimeZoneOffset {declared!r} does not match "
                f"eventTime offset {actual_hhmm!r}"
            )
        return self

DSCSA/GS1 note: EPCIS requires both eventTime and eventTimeZoneOffset on the wire precisely because a UTC-normalized timestamp alone discards the trading partner’s local business context; rejecting a mismatch between the two here — rather than silently trusting one — is what keeps the six-year audit trail internally consistent.

Step 2 — Validate the EPC list against SGTIN URI syntax #

The epcList is the field DSCSA verification actually depends on. Each entry must be a syntactically valid urn:epc:id:sgtin:... URI whose company-prefix, item-reference, and serial segments respect GS1’s General Specifications, not just “any non-empty string.”

SGTIN_RE = re.compile(
    r"^urn:epc:id:sgtin:(?P<prefix>\d{6,12})\.(?P<itemref>\d{1,7})\."
    r"(?P<serial>[A-Za-z0-9\-.\_]{1,20})$"
)


def parse_sgtin(epc: str) -> dict[str, str]:
    """Split a validated SGTIN URI into its GS1 components."""
    match = SGTIN_RE.match(epc)
    if not match:
        raise ValueError(f"not a well-formed SGTIN URI: {epc!r}")
    return match.groupdict()


class ObjectEvent(EPCISEventBase):
    type: Literal["ObjectEvent"]
    action: Literal["ADD", "OBSERVE", "DELETE"]
    epcList: list[str] = Field(min_length=1)

    @field_validator("epcList")
    @classmethod
    def _epcs_are_sgtins(cls, v: list[str]) -> list[str]:
        for epc in v:
            parse_sgtin(epc)  # raises ValueError on malformed syntax
        return v

DSCSA/GS1 note: GTIN (01) and serial (21) together form the SGTIN that identifies a single saleable unit; a syntactically invalid entry here would silently propagate an unverifiable identifier all the way to a trading partner’s verification router, which is exactly the failure this validator exists to prevent.

Step 3 — Add lot and expiry as optional ILMD extensions #

Commissioning events typically carry lot (10) and expiration date (17) in an ilmd (Instance/Lot Master Data) block. Model it as optional so observation and shipping events — which may omit it — still validate, while any event that does include it gets the same syntax check.

class Ilmd(BaseModel):
    lotNumber: str
    itemExpirationDate: str  # YYYY-MM-DD in JSON-LD (vs. YYMMDD on the AI)

    @field_validator("itemExpirationDate")
    @classmethod
    def _expiry_format(cls, v: str) -> str:
        if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", v):
            raise ValueError("itemExpirationDate must be YYYY-MM-DD")
        return v


class ObjectEvent(ObjectEvent):
    ilmd: Ilmd | None = None

DSCSA/GS1 note: re-opening ObjectEvent here is illustrative of incremental modeling; in production, declare ilmd on the class directly. Either way, lot (10) and expiry (17) must round-trip byte-for-byte through this model — never reformatted — because a dispenser reconciling a recall notice matches on the exact printed lot string.

Step 4 — Build the discriminated union across all four event types #

ObjectEvent alone cannot represent a real EPCIS stream, which interleaves commissioning, packing, ownership transfer, and repackaging records. A Pydantic discriminated union on the type field dispatches each payload to the correct model without a manual if/elif chain, and it fails fast if type is missing or unrecognized.

from typing import Annotated, Union


class AggregationEvent(EPCISEventBase):
    type: Literal["AggregationEvent"]
    action: Literal["ADD", "OBSERVE", "DELETE"]
    parentID: str
    childEPCs: list[str] = Field(min_length=1)

    @field_validator("childEPCs")
    @classmethod
    def _children_are_sgtins(cls, v: list[str]) -> list[str]:
        for epc in v:
            parse_sgtin(epc)
        return v


class TransactionEvent(EPCISEventBase):
    type: Literal["TransactionEvent"]
    action: Literal["ADD", "OBSERVE", "DELETE"]
    epcList: list[str] = Field(min_length=1)
    bizTransactionList: list[str] = Field(min_length=1)


class TransformationEvent(EPCISEventBase):
    type: Literal["TransformationEvent"]
    inputEPCList: list[str] = Field(min_length=1)
    outputEPCList: list[str] = Field(min_length=1)


EPCISEvent = Annotated[
    Union[ObjectEvent, AggregationEvent, TransactionEvent, TransformationEvent],
    Field(discriminator="type"),
]


class EPCISDocument(BaseModel):
    eventList: list[EPCISEvent] = Field(min_length=1)

DSCSA/GS1 note: ObjectEvent carries commissioning and decommissioning, AggregationEvent carries case/pallet parent-child relationships, TransactionEvent carries ownership change, and TransformationEvent carries repackaging — mixing them without a discriminator risks silently coercing, say, an aggregation payload into an ObjectEvent shape and dropping parentID without any error.

Step 5 — Route ValidationError to a dead-letter queue #

The model set is only half the job; the consuming loop must treat pydantic.ValidationError as a routing decision, not a crash. TypeAdapter validates a single dict against the union without needing a wrapping document.

import asyncio
from pydantic import TypeAdapter, ValidationError

event_adapter = TypeAdapter(EPCISEvent)


async def process_stream(raw_events, epcis_repo, dlq) -> None:
    async for raw in raw_events:                 # dicts already JSON-decoded
        try:
            event = event_adapter.validate_python(raw.payload)
        except ValidationError as exc:
            await dlq.put({
                "payload": raw.payload,
                "errors": exc.errors(include_url=False),
                "offset": raw.offset,
            })
            continue                             # never persist partial data
        await epcis_repo.commit(event)

DSCSA/GS1 note: exc.errors() returns a structured list of field paths and messages rather than a single string, which is what lets a compliance reviewer see exactly which GS1 rule a payload broke without re-parsing the raw JSON-LD by hand — the same tiered classification described in Schema Validation & Error Handling.

Verification #

Confirm the validator both accepts conformant events and rejects each class of defect before wiring it to live traffic. A table-driven test against TypeAdapter is the fastest way to pin the discriminated union’s behavior:

import pytest
from pydantic import ValidationError

GOOD_EVENT = {
    "type": "ObjectEvent",
    "eventTime": "2026-07-01T10:00:00+00:00",
    "eventTimeZoneOffset": "+00:00",
    "action": "ADD",
    "bizStep": "urn:epcglobal:cbv:bizstep:commissioning",
    "disposition": "urn:epcglobal:cbv:disp:active",
    "readPoint": {"id": "urn:epc:id:sgln:0312345.00000.0"},
    "epcList": ["urn:epc:id:sgtin:0312345.011111.SERIAL001"],
}


def test_valid_object_event_round_trips():
    event = event_adapter.validate_python(GOOD_EVENT)
    assert event.action == "ADD"
    assert event.epcList[0].endswith("SERIAL001")


@pytest.mark.parametrize("field, bad_value", [
    ("eventTimeZoneOffset", "0000"),
    ("bizStep", "commissioning"),
    ("epcList", ["not-a-sgtin"]),
])
def test_rejects_malformed_fields(field, bad_value):
    bad_event = {**GOOD_EVENT, field: bad_value}
    with pytest.raises(ValidationError):
        event_adapter.validate_python(bad_event)


def test_naive_timestamp_rejected():
    bad_event = {**GOOD_EVENT, "eventTime": "2026-07-01T10:00:00"}
    with pytest.raises(ValidationError):
        event_adapter.validate_python(bad_event)

Beyond unit tests, run the validator against a batch of real or staged trading-partner payloads and assert the dead-letter rate stays near zero for known-good producers; a sudden rise usually means an upstream schema drift rather than a validator bug. Finally, inspect a sample of dead-lettered records manually and confirm every errors() entry names a specific field and a human-readable reason — an opaque dead-letter entry is as unauditable as no validation at all.

Gotchas & Edge Cases #

• Z suffix vs. explicit offset. Pydantic parses a trailing Z in eventTime as UTC, but the eventTimeZoneOffset field is a separate string the sender must also populate correctly — the cross-field model_validator in Step 1 exists precisely because these two can silently disagree.
• Leading zeros in the SGTIN prefix. Treat parse_sgtin’s captured groups as strings throughout. Casting the company prefix or item reference to int strips leading zeros and corrupts the GTIN (01) a partner expects to see reconstructed byte-for-byte.
• Literal["ADD", "OBSERVE", "DELETE"] is case-sensitive. Some partner systems send lowercase actions; decide explicitly whether to normalize case in a field_validator before the Literal check runs, rather than let a case mismatch silently 422 every event from one partner.
• Discriminator field absent or misspelled. If a payload omits type entirely, Pydantic raises a union-discriminator error rather than falling through to a default model — treat that as a hard schema-drift signal from the sender, not a bug in your validator.
• TypeAdapter instances are not free to rebuild. Construct event_adapter = TypeAdapter(EPCISEvent) once at module scope; rebuilding it per event re-runs Pydantic’s core-schema compilation and measurably slows a high-throughput ingestion loop.

Related #

• Up to the parent topic: Schema Validation & Error Handling
• Parsing EPCIS XML with Python lxml Efficiently — the equivalent streaming validation problem on the XML side
• Step-by-Step Guide to EPCIS 2.0 Event Formatting — the JSON-LD shape these models validate against
• Serialization Data Ingestion & EPCIS Event Sync — the pipeline this validation gate protects