Mapping IFTMIN transport instructions to typed models #

Mapping IFTMIN transport instructions to typed models turns a UN/EDIFACT firm-booking message into a validated Python object graph that feeds the canonical Bill of Lading (B/L) schema without a single untyped field slipping through. An IFTMIN carries the shipper’s binding instruction to move cargo — parties, itinerary, goods, and Verified Gross Mass — and it is legally and commercially heavier than a manifest: a mistyped NAD party role or an unresolved UN/LOCODE here becomes a mis-routed container, a customs hold, or a rejected booking. This page specifies the pydantic model set, the coercion rules, and the mandatory-versus-conditional gating that convert each IFTMIN segment group into a strict, auditable typed record.

Architecture Alignment #

This task is the transport-instruction counterpart to the segment-level decode work in How to map UN/EDIFACT B/L fields to Python dicts: that page produces a lenient dict, this one projects an IFTMIN into strict, validated models. Both sit inside the Bill of Lading Schema Mapping discipline of the broader Core Maritime Architecture & Taxonomy framework, which owns the canonical schema every carrier format resolves into. IFTMIN differs from an aggregated manifest in intent — it is one shipper’s firm instruction, not a summary of many — so its typing layer must be stricter about mandatory parties and dates than the tolerant summary handling documented in IFCSUM EDI Message Parsing. The contract here is: accept the syntactic drift a real carrier network emits, but refuse to construct a model that violates the message’s structural obligations under UN/EDIFACT.

Prerequisites & Environment Setup #

The mapper targets Python 3.11+ (PEP 604 unions, StrEnum, faster Decimal) and pydantic v2 for strict-mode validation. Install the typing and structured-logging stack:

pip install "pydantic>=2.6" structlog pytest

Structured JSON logging is non-negotiable here: every coercion and every rejected qualifier must be reconstructable during a port state control or customs audit, so print() and unstructured logging strings are unacceptable. Configure structlog once at process start:

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.iftmin")

The mapper reads a handful of environment variables so directory versions and compliance floors are configurable per deployment rather than hard-coded:

This page assumes the raw interchange has already been split into tag-keyed segments by the release-character-aware walk documented in the sibling dict mapper; here we take those elements and type them.

Step-by-step Implementation #

Step 1 — Model segment-group primitives and coercion helpers #

IFTMIN structure is defined by the UN/EDIFACT directory (ISO 9735 syntax), where each segment carries positional data elements identified by four-digit DE codes. Before typing any party or place, pin the controlled vocabularies — transport mode (DE 8067), location function (DE 3227), and party function (DE 3035) — as enums so an unexpected code fails loudly rather than propagating as a raw string. DTM values are coerced through their format qualifier (102 = CCYYMMDD, 203 = CCYYMMDDHHMM).

from __future__ import annotations

import os
from datetime import datetime, timezone
from decimal import Decimal
from enum import StrEnum

import structlog

log = structlog.get_logger("maritime.iftmin")

VGM_MIN_KG: Decimal = Decimal(os.environ.get("VGM_MIN_KG", "1000"))
LOCODE_AGENCY: str = os.environ.get("LOCODE_AGENCY", "6")

_DTM_FORMATS: dict[str, str] = {"102": "%Y%m%d", "203": "%Y%m%d%H%M", "101": "%y%m%d"}


class TransportMode(StrEnum):        # UN/EDIFACT DE 8067
    MARITIME = "1"
    RAIL = "2"
    ROAD = "3"
    AIR = "4"
    INLAND_WATER = "8"


class PartyRole(StrEnum):
    CONSIGNOR = "consignor"
    CONSIGNEE = "consignee"
    NOTIFY = "notify_party"
    CARRIER = "carrier"
    FORWARDER = "forwarder"


class IFTMINStructureError(Exception):
    """A mandatory IFTMIN segment group is absent or malformed."""


class UnknownPartyQualifier(Exception):
    """A NAD DE 3035 qualifier has no mapping — quarantine, do not drop."""


def coerce_dtm(function: str, value: str, fmt: str) -> datetime:
    pattern = _DTM_FORMATS.get(fmt)
    if pattern is None:
        raise IFTMINStructureError(f"unsupported DTM format qualifier {fmt!r}")
    parsed = datetime.strptime(value, pattern).replace(tzinfo=timezone.utc)
    log.info("dtm_coerced", function=function, fmt=fmt, iso=parsed.isoformat())
    return parsed

Step 2 — Type the parties with a NAD qualifier validator #

The NAD segment group is the highest-risk mapping in an IFTMIN: a mis-assigned role silently mis-routes cargo. Party function comes from DE 3035 — CZ consignor, CN consignee, NI notify party, CA carrier, FW freight forwarder — but carriers routinely emit synonyms (SU/SH for the shipper, N1 for a second notify). A model_validator normalizes the qualifier through one authoritative map and raises UnknownPartyQualifier for anything unmapped, so an unrecognised role is quarantined for human reconciliation rather than coerced into the wrong slot or dropped.

from pydantic import BaseModel, ConfigDict, Field, model_validator

NAD_QUALIFIER_MAP: dict[str, PartyRole] = {
    "CZ": PartyRole.CONSIGNOR, "SU": PartyRole.CONSIGNOR, "SH": PartyRole.CONSIGNOR,
    "CN": PartyRole.CONSIGNEE,
    "NI": PartyRole.NOTIFY, "N1": PartyRole.NOTIFY,
    "CA": PartyRole.CARRIER,
    "FW": PartyRole.FORWARDER,
}


class Party(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, frozen=True)

    qualifier: str = Field(..., min_length=2, max_length=3)
    party_id: str | None = None          # C082.3039 party identification
    name: str | None = None              # C080 party name
    role: PartyRole | None = None

    @model_validator(mode="after")
    def resolve_role(self) -> "Party":
        role = NAD_QUALIFIER_MAP.get(self.qualifier.upper())
        if role is None:
            log.warning("nad_qualifier_unmapped", qualifier=self.qualifier)
            raise UnknownPartyQualifier(self.qualifier)
        object.__setattr__(self, "role", role)
        log.info("nad_role_resolved", qualifier=self.qualifier, role=role.value)
        return self

Step 3 — Type the itinerary with a UN/LOCODE validator and TDT leg #

LOC places (DE 3227: 9 port of loading, 11 port of discharge, 13 transhipment, 5 departure, 8 destination) carry a UN/LOCODE in composite C517, together with a code-list responsible agency in DE 3055 where 6 denotes UN/ECE. A field_validator enforces the UN/LOCODE grammar — two ISO 3166-1 alpha-2 country letters followed by a three-character location code drawn from A–Z and 2–9 — and flags any non-UN/ECE agency without hard-rejecting, because a carrier using a proprietary code list is a reconciliation case, not corruption. The TDT group binds the transport mode (DE 8067) and carrier identity (composite C040).

import re

from pydantic import field_validator

_LOCODE_RE = re.compile(r"^[A-Z]{2}[A-Z2-9]{3}$")
ISO_3166_ALPHA2: frozenset[str] = frozenset({"US", "NL", "DE", "SG", "CN", "GB", "AE", "BR"})


class LocationFunction(StrEnum):     # DE 3227
    PORT_OF_LOADING = "9"
    PORT_OF_DISCHARGE = "11"
    TRANSHIPMENT = "13"
    DEPARTURE = "5"
    DESTINATION = "8"


class Location(BaseModel):
    function: LocationFunction
    un_locode: str
    agency: str = LOCODE_AGENCY
    name: str | None = None

    @field_validator("un_locode")
    @classmethod
    def validate_locode(cls, v: str) -> str:
        v = v.strip().upper()
        if not _LOCODE_RE.fullmatch(v):
            raise ValueError(f"malformed UN/LOCODE {v!r}")
        if v[:2] not in ISO_3166_ALPHA2:
            raise ValueError(f"unknown ISO 3166 country prefix {v[:2]!r}")
        return v

    @field_validator("agency")
    @classmethod
    def check_agency(cls, v: str) -> str:
        if v != LOCODE_AGENCY:
            log.warning("locode_agency_nonstandard", agency=v, expected=LOCODE_AGENCY)
        return v


class TransportLeg(BaseModel):
    stage: str                            # DE 8051 (20 = main-carriage)
    mode: TransportMode                   # DE 8067
    conveyance_ref: str | None = None     # C222.8212 voyage number
    carrier_id: str | None = None         # C040.3127
    departure: Location | None = None
    arrival: Location | None = None

Step 4 — Type goods, packaging, and the SOLAS VGM measurement #

The goods segment group binds GID (goods item number, package count/type in composite C213), PAC (outer packaging), free-text FTX descriptions, and MEA measurements. MEA is where SOLAS bites: a Verified Gross Mass arrives as MEA+AAE+VGM+KGM:21000 — purpose qualifier DE 6311, measured attribute DE 6313 (VGM), unit DE 6411 (KGM), value. A field_validator rejects any VGM below the plausibility floor so an implausible or tare-only weight never reaches the stowage planner.

class Package(BaseModel):
    count: int = Field(..., ge=0)
    type_code: str | None = None          # DE 7065 package type


class Measurement(BaseModel):
    purpose: str                          # DE 6311 (AAE, WT)
    attribute: str                        # DE 6313 (VGM, AAB)
    unit: str                             # DE 6411 (KGM)
    value: Decimal = Field(..., ge=0)


class GoodsItem(BaseModel):
    item_number: int = Field(..., ge=1)
    packages: list[Package] = Field(default_factory=list)
    description: str | None = None        # FTX free text
    verified_gross_mass: Decimal | None = None

    @field_validator("verified_gross_mass")
    @classmethod
    def check_vgm(cls, v: Decimal | None) -> Decimal | None:
        if v is not None and v < VGM_MIN_KG:
            raise ValueError(f"VGM {v} below SOLAS floor {VGM_MIN_KG}")
        return v

Step 5 — Assemble the instruction and gate mandatory-vs-conditional groups #

The top-level TransportInstruction composes the typed groups and enforces IFTMIN’s structural obligations. BGM (message function DE 1225, document name code DE 1001) and at least a consignor and consignee NAD are mandatory; a transhipment LOC, the requested-delivery DTM, and FTX remarks are conditional and default cleanly to None or empty. A model_validator distinguishes a format failure (missing mandatory group) from a business exception, so the caller can route the former to the dead-letter queue and the latter to quarantine — the same split the parent Bill of Lading Schema Mapping layer relies on.

class TransportInstruction(BaseModel):
    booking_reference: str = Field(..., min_length=1)   # BGM C106.1004
    message_function: str                               # BGM DE 1225 (9 = original)
    document_name_code: str | None = None               # BGM C002.1001
    parties: list[Party]
    itinerary: list[Location]
    transport_legs: list[TransportLeg] = Field(default_factory=list)
    goods: list[GoodsItem]
    requested_delivery: datetime | None = None          # conditional DTM
    remarks: list[str] = Field(default_factory=list)    # conditional FTX

    @model_validator(mode="after")
    def enforce_mandatory_groups(self) -> "TransportInstruction":
        roles = {p.role for p in self.parties}
        missing = {PartyRole.CONSIGNOR, PartyRole.CONSIGNEE} - roles
        if missing:
            raise IFTMINStructureError(f"mandatory NAD role(s) absent: {sorted(missing)}")
        if not self.goods:
            raise IFTMINStructureError("IFTMIN requires at least one goods item")
        if not any(loc.function == LocationFunction.PORT_OF_LOADING for loc in self.itinerary):
            log.warning("loc_pol_absent", booking=self.booking_reference)
        log.info(
            "iftmin_typed",
            booking=self.booking_reference,
            parties=len(self.parties),
            goods=len(self.goods),
            legs=len(self.transport_legs),
        )
        return self

Edge Cases & Carrier Deviations #

The most common non-fatal deviation is a non-standard NAD qualifier: a forwarder emits NAD+FW where a handler expects the consignor under NAD+CZ, or a second notify party under NAD+N1. Because the document is legally valid, treat an unmapped qualifier as a quarantine reason — capture the raw code in the structured log, let ops extend the map, and replay — rather than a hard drop that discards a firm booking a human could reconcile in seconds.

Verification & Testing #

Assert the two behaviours that most often regress — NAD role normalization and the UN/LOCODE validator — plus the mandatory-group gate, against fixed element fixtures with pytest. Use structlog’s capture helper to prove the audit event fires.

import pytest
import structlog
from pydantic import ValidationError

from maritime_iftmin.models import (
    GoodsItem, Location, LocationFunction, Party, PartyRole,
    TransportInstruction, UnknownPartyQualifier,
)


def test_nad_synonym_normalises_to_consignor() -> None:
    assert Party(qualifier="SU", name="ACME SHIPPING").role is PartyRole.CONSIGNOR


def test_unknown_nad_qualifier_raises_for_quarantine() -> None:
    with pytest.raises(UnknownPartyQualifier):
        Party(qualifier="ZZ", name="MYSTERY PARTY")


def test_locode_validator_rejects_bad_country() -> None:
    with pytest.raises(ValidationError):
        Location(function=LocationFunction.PORT_OF_LOADING, un_locode="XXNYC")


def test_missing_consignor_fails_structure() -> None:
    with pytest.raises(ValidationError):
        TransportInstruction(
            booking_reference="MAEU123", message_function="9",
            parties=[Party(qualifier="CN", name="GLOBAL IMPORTERS")],
            itinerary=[Location(function=LocationFunction.PORT_OF_LOADING, un_locode="USNYC")],
            goods=[GoodsItem(item_number=1)],
        )


def test_typed_event_is_emitted() -> None:
    cap = structlog.testing.LogCapture()
    structlog.configure(processors=[cap])
    TransportInstruction(
        booking_reference="MAEU123", message_function="9",
        parties=[Party(qualifier="CZ"), Party(qualifier="CN")],
        itinerary=[Location(function=LocationFunction.PORT_OF_LOADING, un_locode="USNYC")],
        goods=[GoodsItem(item_number=1, verified_gross_mass=21000)],
    )
    assert cap.entries[-1]["event"] == "iftmin_typed"
    assert cap.entries[-1]["goods"] == 1

A passing run emits one JSON line per typed instruction: {"event": "iftmin_typed", "booking": "MAEU123", "parties": 2, "goods": 1, "legs": 0, "log_level": "info", "timestamp": "..."}. Assert on the structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• How to map UN/EDIFACT B/L fields to Python dicts — the release-character-aware decode this typing layer consumes
• Bill of Lading Schema Mapping — the canonical schema and DLQ-versus-quarantine routing these models feed
• IFCSUM EDI Message Parsing — the aggregated-manifest counterpart reconciled against typed instructions
• Container Hierarchy Data Models — where the typed goods items resolve into the equipment topology
• Core Maritime Architecture & Taxonomy — the framework governing schema, naming, and lifecycle state

Up: Bill of Lading Schema Mapping — the parent discipline that types and validates every carrier format into one record.