Handling UNA service strings and delimiter drift #

Handling UNA service strings and delimiter drift is the difference between a parser that survives every carrier feed and one that silently collapses an entire UN/EDIFACT interchange into a single unparseable segment the moment a partner ships non-default delimiters. The optional UNA service string advice at the head of an interchange declares the six characters that punctuate every segment, element, and component that follows; ignore it and hard-code the familiar +, :, and ' and the first carrier that redefines even one of them turns your whole manifest into noise. This page specifies a typed reader for the service string, a release-character-aware tokenizer, and a drift detector that flags a delimiter change before it corrupts a vessel-clearance cycle.

Architecture Alignment #

Delimiter resolution is the first byte-level decision inside IFCSUM EDI Message Parsing, the discipline that turns an aggregated UN/EDIFACT manifest into a typed record, and it sits under the broader Document Ingestion & EDI Parsing Workflows domain that governs how every wire format enters the platform. This page owns exactly one contract: given raw interchange bytes, produce an authoritative delimiter set and a release-aware split so that everything downstream can trust segment and element boundaries. The streaming segment loop and mandatory-segment gate live in Parsing IFCSUM 311 messages with Python; the structural and semantic checks that a correctly tokenized record must clear live in Schema Validation Frameworks. Getting the delimiter table wrong upstream makes every one of those layers report phantom errors, so this is the cheapest place to fail fast and the most expensive place to guess.

Prerequisites & Environment Setup #

The resolver is pure Python plus structlog; it decodes bytes and never touches the network, which keeps it trivially unit-testable against raw fixtures. Target Python 3.11+ for slots=True on frozen dataclasses and precise str slicing semantics.

python -m venv .venv && source .venv/bin/activate
pip install "structlog>=24.1" "pytest>=8.0"

Structured JSON logging is mandatory here: a delimiter decision is an audit-relevant event because it determines how every subsequent byte is interpreted, and a customs or port-authority review must be able to reconstruct which delimiter set was applied to a given interchange control reference. Configure structlog once at import time so every module emits the same JSON shape:

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("edifact.service_string")

The resolver reads a small amount of configuration so a per-partner baseline and drift threshold are diffable across releases rather than buried in code:

Step-by-step Implementation #

Each step is runnable in isolation and composes into a single delimiter-resolution front end for the IFCSUM parser. The syntax rules referenced throughout are defined by ISO 9735, the standard that specifies the UN/EDIFACT service string advice, the six service characters, and the release mechanism.

Step 1 — Model the service string as a typed ServiceStringAdvice #

The UNA segment is exactly nine characters: the literal tag UNA followed by six positional service characters, in a fixed order — component data element separator, data element separator, decimal notation, release character, a reserved position (which must hold a space), and the segment terminator. Model those six as an immutable, typed record so the delimiter set can never be mutated halfway through a parse and so an invalid combination fails at construction rather than deep inside tokenisation. Implicit dictionaries of loose strings are unacceptable when the values decide how every downstream byte is framed.

from __future__ import annotations

from dataclasses import dataclass

import structlog

log = structlog.get_logger("edifact.service_string")

# ISO 9735 default service characters, applied when no UNA advice is present.
DEFAULT_COMPONENT = ":"
DEFAULT_ELEMENT = "+"
DEFAULT_DECIMAL = "."
DEFAULT_RELEASE = "?"
DEFAULT_SEGMENT = "'"


@dataclass(frozen=True, slots=True)
class ServiceStringAdvice:
    component: str  # position 1 — component data element separator
    element: str    # position 2 — data element separator
    decimal: str    # position 3 — decimal notation character
    release: str    # position 4 — release (escape) character
    reserved: str   # position 5 — reserved, must be a space per ISO 9735
    segment: str    # position 6 — segment terminator

    def as_dict(self) -> dict[str, str]:
        return {
            "component": self.component,
            "element": self.element,
            "decimal": self.decimal,
            "release": self.release,
            "segment": self.segment,
        }

Step 2 — Parse UNA or fall back to the ISO 9735 defaults #

An interchange either opens with UNA and declares its own characters, or omits it and implicitly agrees to the ISO 9735 defaults (:, +, ., ?, space, '). Both paths must yield the same typed object so the rest of the pipeline is delimiter-agnostic. Crucially, the UNA bytes must only be consumed when the payload literally starts with UNA; a stray probe that slices nine bytes off a UNB-led interchange would silently amputate the envelope header. Validation also rejects a degenerate advice — a delimiter reused across two positions makes the syntax ambiguous and must be caught before a single segment is split.

class ServiceStringError(ValueError):
    """Raised when a UNA advice is truncated or internally ambiguous."""


def defaults(decimal: str = DEFAULT_DECIMAL) -> ServiceStringAdvice:
    return ServiceStringAdvice(
        component=DEFAULT_COMPONENT,
        element=DEFAULT_ELEMENT,
        decimal=decimal,
        release=DEFAULT_RELEASE,
        reserved=" ",
        segment=DEFAULT_SEGMENT,
    )


def read_service_string(raw: str, *, strict_reserved: bool = True) -> ServiceStringAdvice:
    if not raw.startswith("UNA"):
        advice = defaults()
        log.info("una_absent", applied="iso_9735_defaults", **advice.as_dict())
        return advice
    if len(raw) < 9:
        raise ServiceStringError("UNA advice truncated: six service characters required")

    chars = raw[3:9]
    advice = ServiceStringAdvice(
        component=chars[0], element=chars[1], decimal=chars[2],
        release=chars[3], reserved=chars[4], segment=chars[5],
    )

    delimiters = [advice.component, advice.element, advice.release, advice.segment]
    if len(set(delimiters)) != len(delimiters):
        raise ServiceStringError(f"UNA delimiters not distinct: {delimiters!r}")
    if strict_reserved and advice.reserved != " ":
        raise ServiceStringError(f"UNA reserved position must be a space, got {advice.reserved!r}")

    log.info("una_parsed", **advice.as_dict())
    return advice

Step 3 — Split with a release-character-aware tokenizer #

The release character (default ?) is EDIFACT’s escape mechanism: it makes the next character literal data rather than a delimiter, so a consignee name containing Smith + Sons is transmitted as Smith ?+ Sons and a free-text field ending in an apostrophe carries ?' rather than a premature segment terminator. A naive str.split() on the delimiter is therefore wrong twice over — it breaks on escaped delimiters that belong to the data, and it cannot tell an escaped terminator from a real one. The single reusable primitive below scans character by character, consuming the character after a release verbatim, and every segment and element split routes through it so the escape rule is applied identically everywhere.

def split_on(raw: str, delimiter: str, release: str) -> list[str]:
    """Split raw on unescaped occurrences of delimiter, honouring the release char."""
    fields: list[str] = []
    buffer: list[str] = []
    i, n = 0, len(raw)
    while i < n:
        ch = raw[i]
        if ch == release and i + 1 < n:
            buffer.append(raw[i + 1])  # released: take the next character literally
            i += 2
            continue
        if ch == delimiter:
            fields.append("".join(buffer))
            buffer = []
            i += 1
            continue
        buffer.append(ch)
        i += 1
    fields.append("".join(buffer))
    return fields


def tokenize_interchange(body: str, advice: ServiceStringAdvice) -> list[list[str]]:
    """Return an ordered list of segments, each a list of elements, escape-aware."""
    segments: list[list[str]] = []
    for raw_segment in split_on(body, advice.segment, advice.release):
        stripped = raw_segment.strip()
        if not stripped:
            continue
        elements = split_on(stripped, advice.element, advice.release)
        segments.append(elements)
    log.info("interchange_tokenised", segment_count=len(segments))
    return segments

Step 4 — Detect delimiter drift before it corrupts a feed #

Delimiter drift is when a trading partner starts emitting a different service-character set than the one your integration was built and tested against — a migration from one EDI translator to another, a re-configured VAN, or a mistyped UNA in a hand-built test file. Two signals catch it. First, compare the resolved advice against a per-sender baseline: any diverging position is a change worth alerting on even when parsing still succeeds, because a shifted decimal character quietly mis-reads a Verified Gross Mass. Second, watch the segment count — when the wrong terminator is assumed, the whole interchange collapses into one giant segment, so a near-zero segment count on a non-trivial payload is the unmistakable fingerprint of a hard-coded delimiter meeting a drifted feed.

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class DriftReport:
    drifted: bool
    diverging_fields: tuple[str, ...]
    collapsed: bool
    segment_count: int


def detect_delimiter_drift(
    advice: ServiceStringAdvice,
    baseline: ServiceStringAdvice,
    segment_count: int,
    *,
    min_expected_segments: int = 3,
) -> DriftReport:
    diverging = tuple(
        name
        for name in ("component", "element", "decimal", "release", "segment")
        if getattr(advice, name) != getattr(baseline, name)
    )
    collapsed = segment_count < min_expected_segments
    drifted = bool(diverging) or collapsed
    if drifted:
        log.warning(
            "delimiter_drift_detected",
            diverging=list(diverging),
            collapsed=collapsed,
            segment_count=segment_count,
            expected=baseline.as_dict(),
            observed=advice.as_dict(),
        )
    return DriftReport(drifted, diverging, collapsed, segment_count)

Edge Cases & Carrier Deviations #

The most instructive deviation is the silently absent UNA. Many forwarders omit the service string entirely and rely on the ISO 9735 defaults, which is perfectly valid — but a brittle probe that unconditionally strips nine bytes will eat the first characters of the UNB envelope and produce a cascade of spurious structural failures downstream. Consuming service characters only when the payload begins with the literal UNA tag is what lets a default-delimiter feed and a UNA-declared feed flow through the identical code path, and it is the single guard that keeps an absent header from masquerading as delimiter corruption.

Verification & Testing #

Assert the three behaviours that break real feeds: non-default delimiters are honoured, escaped delimiters survive tokenisation as literal data, and an absent UNA cleanly falls back to defaults. Drive the resolver end to end with pytest fixtures over raw strings.

import pytest

from edifact.service_string import (
    ServiceStringAdvice,
    defaults,
    detect_delimiter_drift,
    read_service_string,
    split_on,
    tokenize_interchange,
)

# A carrier interchange that redefines the component and terminator characters.
DRIFTED = "UNA|*.? ~UNB|UNOA*2~UNH|1|IFCSUM*D*03B*UN~BGM|785~UNT|3|1~"


def test_una_overrides_defaults() -> None:
    advice = read_service_string(DRIFTED)
    assert advice.element == "|"
    assert advice.component == "*"
    assert advice.segment == "~"


def test_absent_una_uses_iso_9735_defaults() -> None:
    advice = read_service_string("UNB+UNOA:2+FWDR:ZZ'UNH+1+IFCSUM:D:03B:UN'")
    assert advice == defaults()


def test_release_char_keeps_escaped_delimiter_literal() -> None:
    advice = defaults()
    fields = split_on("Smith ?+ Sons+Rotterdam", advice.element, advice.release)
    assert fields == ["Smith + Sons", "Rotterdam"]


def test_drift_detects_collapsed_interchange() -> None:
    # A feed parsed with the wrong terminator yields a single segment.
    body = "UNB+UNOA:2'UNH+1+IFCSUM:D:03B:UN'BGM+785'"
    collapsed = tokenize_interchange(body, ServiceStringAdvice(":", "+", ".", "?", " ", "~"))
    report = detect_delimiter_drift(
        ServiceStringAdvice(":", "+", ".", "?", " ", "~"), defaults(), len(collapsed)
    )
    assert report.collapsed is True
    assert "segment" in report.diverging_fields

A passing run emits one JSON line per resolution decision — for example {"event": "una_parsed", "component": "*", "element": "|", "decimal": ".", "release": "?", "segment": "~", "level": "info", "timestamp": "..."} — and a warning line the moment drift is detected. Assert on those structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Parsing IFCSUM 311 messages with Python — the streaming segment loop and compliance gate this resolver feeds
• IFCSUM EDI Message Parsing — the discipline that maps tokenised segments into typed models
• Schema Validation Frameworks — the structural and semantic checks a correctly tokenised record must clear
• Document Ingestion & EDI Parsing Workflows — the parent domain governing every wire format entering the platform

Up: IFCSUM EDI Message Parsing — the discipline that owns envelope validation, delimiter resolution, and typed model mapping.