Pydantic envelope validation for UN/EDIFACT interchanges #

Every UN/EDIFACT transmission a maritime gateway receives is wrapped in a service envelope — a UNB interchange header, one or more UNH…UNT message envelopes, and a closing UNZ trailer — whose control counters must reconcile before a single business segment is trusted. Pydantic envelope validation for UN/EDIFACT interchanges enforces that reconciliation as a hard gate: it models the ISO 9735 envelope as typed pydantic v2 objects and refuses to hand a payload to the message-body parser until UNB.0020 equals UNZ.0020, the UNZ.0036 interchange control count equals the number of messages actually present, each UNH.0062 equals its matching UNT.0062, and UNT.0074 equals the real segment count. A truncated or spliced interchange fails here, at construction, instead of corrupting a stowage plan three hops downstream.

Architecture Alignment #

This page is the envelope-integrity tier of the Schema Validation Frameworks reference, itself part of the broader Document Ingestion & EDI Parsing Workflows domain. Its responsibility is deliberately narrow: prove the transmission is a well-formed, self-consistent interchange before any field-level or regulatory rule runs. The envelope check is the structural tier’s first move — a UNZ.0036 mismatch is transmission corruption, not a business exception, so it fails fast rather than routing to quarantine. Message-body extraction, once the envelope clears, defers to IFCSUM EDI Message Parsing, and the choice to build these models on pydantic v2’s validator API rather than the v1 root-validator idiom is argued in pydantic v1 vs v2 for maritime schema validation. Keeping envelope validation separate from body parsing is what lets a single malformed control count reject 40 kB of segments in microseconds.

Prerequisites & Environment Setup #

The validator targets Python 3.11+ for Self-typed classmethods and precise exception groups. Install pydantic v2 and structured logging:

pip install "pydantic>=2.6,<3" structlog

Structured JSON logging is mandatory here: an envelope rejection must be replayable during an IMO FAL or customs audit, so every counter mismatch is emitted as a queryable event rather than a print() string. Configure structlog once, at process boot:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("maritime.edifact.envelope")

The parser reads its delimiters and limits from the environment so a carrier that ships a non-standard UNA service string does not require a code change:

Read these delimiters from the UNA service string when it is present and fall back to the ISO 9735 UNOA defaults above only when it is absent — never assume ' unconditionally, or a carrier terminating segments with a different byte collapses the whole interchange into one segment.

Step-by-step Implementation #

Step 1 — Model the UNB interchange header and its composites #

The UNB segment opens every interchange and carries four composites defined by ISO 9735: S001 (syntax identifier 0001 and version 0002), S002 (sender identification 0004), S003 (recipient identification 0010), and S004 (date and time of preparation, 0017/0019). The interchange control reference 0020 is the value that must later match UNZ.0020. Each composite becomes a typed pydantic model so a missing syntax version or a malformed sender qualifier fails at construction, not downstream.

from __future__ import annotations

from pydantic import BaseModel, Field, field_validator

import structlog

log = structlog.get_logger("maritime.edifact.envelope")


class S001SyntaxIdentifier(BaseModel):
    syntax_id: str = Field(..., min_length=4, max_length=4)   # 0001, e.g. UNOA
    syntax_version: str = Field(..., min_length=1, max_length=1)  # 0002

    @field_validator("syntax_id")
    @classmethod
    def known_repertoire(cls, v: str) -> str:
        if v not in {"UNOA", "UNOB", "UNOC", "UNOD", "UNOY"}:
            raise ValueError(f"unsupported syntax identifier S001.0001={v!r}")
        return v


class S002InterchangeSender(BaseModel):
    sender_id: str = Field(..., min_length=1, max_length=35)   # 0004
    partner_qualifier: str | None = Field(default=None, max_length=4)  # 0007


class S003InterchangeRecipient(BaseModel):
    recipient_id: str = Field(..., min_length=1, max_length=35)  # 0010
    partner_qualifier: str | None = Field(default=None, max_length=4)


class S004DateTime(BaseModel):
    prepared_date: str = Field(..., pattern=r"^\d{6,8}$")  # 0017 YYMMDD/CCYYMMDD
    prepared_time: str = Field(..., pattern=r"^\d{4}$")     # 0019 HHMM


class UNBHeader(BaseModel):
    syntax: S001SyntaxIdentifier
    sender: S002InterchangeSender
    recipient: S003InterchangeRecipient
    prepared: S004DateTime
    control_ref: str = Field(..., min_length=1, max_length=14)  # 0020

Step 2 — Model UNH/UNT and enforce the message-level equalities #

Each message is bounded by UNH (message reference 0062 and the S009 message identifier — type IFCSUM, version D, release 96A, agency UN) and UNT (segment count 0074 and a repeat of the message reference 0062). Two invariants live at this level: UNH.0062 must equal UNT.0062, and UNT.0074 must equal the number of segments in the message inclusive of the UNH and UNT segments themselves — a rule engineers routinely get off by two. A model_validator(mode="after") cross-checks both once the whole message object exists.

from typing import Self

from pydantic import model_validator


class S009MessageIdentifier(BaseModel):
    message_type: str = Field(..., max_length=6)   # 0065, e.g. IFCSUM
    version: str = Field(..., max_length=3)         # 0052, e.g. D
    release: str = Field(..., max_length=3)         # 0054, e.g. 96A
    agency: str = Field(..., max_length=3)          # 0051, e.g. UN


class EDIFACTMessage(BaseModel):
    unh_ref: str = Field(..., max_length=14)        # UNH 0062
    identifier: S009MessageIdentifier
    body_segment_count: int = Field(..., ge=0)      # segments between UNH and UNT
    unt_segment_count: int = Field(..., ge=2)       # UNT 0074
    unt_ref: str = Field(..., max_length=14)        # UNT 0062

    @model_validator(mode="after")
    def check_message_control(self) -> Self:
        if self.unh_ref != self.unt_ref:
            log.error("message_ref_mismatch", unh_0062=self.unh_ref, unt_0062=self.unt_ref)
            raise ValueError(f"UNH.0062={self.unh_ref} != UNT.0062={self.unt_ref}")
        declared = self.unt_segment_count
        actual = self.body_segment_count + 2  # + UNH + UNT
        if declared != actual:
            log.error("segment_count_mismatch", unt_0074=declared, computed=actual)
            raise ValueError(f"UNT.0074={declared} != actual segment count {actual}")
        return self

Step 3 — Model the interchange and enforce the interchange-level equalities #

The Interchange model composes the UNB header, the list of validated messages, and the UNZ trailer (0036 interchange control count and 0020 control reference). Its model_validator enforces the two envelope-spanning rules from ISO 9735: UNB.0020 == UNZ.0020 proves the header and trailer belong to the same transmission, and UNZ.0036 == len(messages) proves no message was lost or duplicated in transit. Both failures are fatal — a control-count drift is corruption, so it is raised, never silently repaired.

class UNZTrailer(BaseModel):
    control_count: int = Field(..., ge=0)           # 0036
    control_ref: str = Field(..., max_length=14)    # 0020


class Interchange(BaseModel):
    header: UNBHeader
    messages: list[EDIFACTMessage]
    trailer: UNZTrailer

    @model_validator(mode="after")
    def check_interchange_control(self) -> Self:
        if self.header.control_ref != self.trailer.control_ref:
            log.error(
                "interchange_ref_mismatch",
                unb_0020=self.header.control_ref,
                unz_0020=self.trailer.control_ref,
            )
            raise ValueError("UNB.0020 does not equal UNZ.0020")
        if self.trailer.control_count != len(self.messages):
            log.error(
                "interchange_count_mismatch",
                unz_0036=self.trailer.control_count,
                messages_present=len(self.messages),
            )
            raise ValueError(
                f"UNZ.0036={self.trailer.control_count} != {len(self.messages)} messages"
            )
        refs = [m.unh_ref for m in self.messages]
        if len(set(refs)) != len(refs):
            raise ValueError("duplicate UNH.0062 message reference within interchange")
        log.info(
            "interchange_validated",
            control_ref=self.header.control_ref,
            message_count=len(self.messages),
        )
        return self

Step 4 — Tokenize the raw envelope and validate before body parsing #

The final piece reads delimiters from the UNA service string (or the ISO 9735 defaults), splits the raw bytes into segments, and populates the models from only the service segments — UNB, UNH, UNT, UNZ. The message body between each UNH and UNT is counted here but deliberately not parsed: body extraction is the responsibility of the IFCSUM EDI Message Parsing stage, invoked only after this function returns a validated Interchange.

import os


def _delimiters(raw: str) -> tuple[str, str, str]:
    if raw.startswith("UNA") and len(raw) >= 9:
        return raw[4], raw[3], raw[6]  # segment term, element sep, component sep
    return (
        os.environ.get("EDIFACT_SEGMENT_TERM", "'"),
        os.environ.get("EDIFACT_ELEMENT_SEP", "+"),
        os.environ.get("EDIFACT_COMPONENT_SEP", ":"),
    )


def validate_envelope(raw: str) -> Interchange:
    seg_term, elem_sep, comp_sep = _delimiters(raw)
    body = raw[9:] if raw.startswith("UNA") else raw
    segments = [s.strip() for s in body.split(seg_term) if s.strip()]

    header: UNBHeader | None = None
    trailer: UNZTrailer | None = None
    messages: list[EDIFACTMessage] = []
    open_unh: dict[str, str] | None = None
    body_count = 0

    for seg in segments:
        tag = seg[:3]
        el = seg.split(elem_sep)
        if tag == "UNB":
            s001, s002, s003, s004 = (el[1].split(comp_sep), el[2].split(comp_sep),
                                      el[3].split(comp_sep), el[4].split(comp_sep))
            header = UNBHeader(
                syntax=S001SyntaxIdentifier(syntax_id=s001[0], syntax_version=s001[1]),
                sender=S002InterchangeSender(sender_id=s002[0]),
                recipient=S003InterchangeRecipient(recipient_id=s003[0]),
                prepared=S004DateTime(prepared_date=s004[0], prepared_time=s004[1]),
                control_ref=el[5],
            )
        elif tag == "UNH":
            s009 = el[2].split(comp_sep)
            open_unh = {"ref": el[1], "type": s009[0], "ver": s009[1],
                        "rel": s009[2], "agy": s009[3]}
            body_count = 0
        elif tag == "UNT":
            assert open_unh is not None, "UNT before UNH"
            messages.append(EDIFACTMessage(
                unh_ref=open_unh["ref"],
                identifier=S009MessageIdentifier(
                    message_type=open_unh["type"], version=open_unh["ver"],
                    release=open_unh["rel"], agency=open_unh["agy"]),
                body_segment_count=body_count,
                unt_segment_count=int(el[1]),
                unt_ref=el[2],
            ))
            open_unh = None
        elif tag == "UNZ":
            trailer = UNZTrailer(control_count=int(el[1]), control_ref=el[2])
        elif open_unh is not None:
            body_count += 1  # a body segment inside the current message

    if header is None or trailer is None:
        raise ValueError("interchange missing UNB header or UNZ trailer")
    return Interchange(header=header, messages=messages, trailer=trailer)

Edge Cases & Carrier Deviations #

A recurring trap is the missing UNA service string combined with functional-group nesting. When a carrier wraps messages in UNG/UNE groups, ISO 9735 says UNZ.0036 counts the groups, not the individual messages — so a flat len(messages) check rejects a perfectly legal interchange. Detect UNG during tokenization and switch the UNZ.0036 comparison to the group count; only fall back to the message count for the common ungrouped case shown above.

Verification & Testing #

Assert both the happy path and each counter violation with pytest fixtures built from raw interchange strings, so the delimiter tokenizer and every model_validator are exercised end to end.

import pytest
from pydantic import ValidationError

from maritime_edifact.envelope import validate_envelope


@pytest.fixture
def good_interchange() -> str:
    return (
        "UNB+UNOA:2+SENDER+RECIPIENT+260717:0900+IC0001'"
        "UNH+M1+IFCSUM:D:96A:UN'BGM+85'NAD+CA+MAEU'UNT+4+M1'"
        "UNZ+1+IC0001'"
    )


def test_valid_envelope_passes(good_interchange: str) -> None:
    ix = validate_envelope(good_interchange)
    assert ix.trailer.control_count == len(ix.messages) == 1
    assert ix.header.control_ref == ix.trailer.control_ref == "IC0001"


def test_interchange_ref_mismatch_rejected(good_interchange: str) -> None:
    corrupt = good_interchange.replace("UNZ+1+IC0001", "UNZ+1+IC9999")
    with pytest.raises(ValidationError, match="UNB.0020 does not equal UNZ.0020"):
        validate_envelope(corrupt)


def test_message_count_mismatch_rejected(good_interchange: str) -> None:
    corrupt = good_interchange.replace("UNZ+1+IC0001", "UNZ+2+IC0001")
    with pytest.raises(ValidationError, match="!= 1 messages"):
        validate_envelope(corrupt)


def test_segment_count_mismatch_rejected(good_interchange: str) -> None:
    corrupt = good_interchange.replace("UNT+4+M1", "UNT+5+M1")
    with pytest.raises(ValidationError, match="UNT.0074"):
        validate_envelope(corrupt)

A rejected interchange emits exactly one structured line before the exception surfaces — for example {"event": "interchange_count_mismatch", "unz_0036": 2, "messages_present": 1, "log_level": "error", "timestamp": "..."}. Assert on those keys, never on a formatted message string, so the audit pipeline stays queryable.

Frequently Asked Questions #

Related #

• Schema Validation Frameworks — the tiered validation gate this envelope check is the structural first move of
• IFCSUM EDI Message Parsing — the message-body extraction that runs only after the envelope validates
• pydantic v1 vs v2 for maritime schema validation — why these models use v2 model_validator cross-checks over v1 root validators
• Document Ingestion & EDI Parsing Workflows — the parent domain governing ingestion, parsing, and validation contracts

Up: Schema Validation Frameworks — the reference that owns the rule engine judging each maritime payload.