Schema Validation Frameworks #

Schema Validation Frameworks are the deterministic rule engine that decides — synchronously, at the normalization boundary — whether an incoming maritime payload is structurally sound, semantically legal, and regulatory-compliant before it consumes any downstream compute. Every container movement, customs declaration, and terminal handoff rides on structured payloads that must conform to strict commercial and regulatory contracts, whether they arrive as UN/EDIFACT transmissions, OCR-extracted PDF fields, or terminal REST responses. This reference sits inside the Document Ingestion & EDI Parsing Workflows framework and owns one job: reject or quarantine malformed records at the first gate, so a misparsed weight or a drifting check digit never reaches a stowage planner, a customs gateway, or a berth plan. For shipping ops teams and Python automation engineers, a correctly deployed validation framework eliminates manual reconciliation, prevents terminal-system rejections, and produces the immutable audit trail that port-state control and customs inspections demand.

Ingestion Boundary & Protocol Handling #

Validation is embedded synchronously at the normalization boundary — the moment a raw interchange has been decoded into a candidate record — never deferred to a post-processing job where a bad payload has already fanned out. Inputs reach this boundary over the same transport families the rest of the parent workflow handles: AS2/OFTP2 EDI VAN interchanges carrying UN/EDIFACT IFCSUM, IFTMIN, and VERMAS messages; SFTP drops of delimited manifests; HTTPS webhooks from carrier and terminal REST APIs; and OCR output promoted from PDF Bill of Lading Extraction. Each source arrives with a different fidelity profile, so the framework applies a syntax-tolerant, semantics-strict posture: it accepts minor transport-level noise but refuses to let a record cross the gate until it matches an explicit contract.

The first check at the boundary is version selection. Terminal operators bump API versions, carriers rotate EDIFACT control-number sequences, and OCR models degrade under new scanner hardware, so the framework routes each payload to the correct schema model by inspecting its declared version header (UNH message-type/version for EDIFACT, an API version tag for REST, a template ID for OCR). This is what keeps a D.95B IFCSUM and a D.16A IFCSUM from being judged against the same field rules. Only once the version is resolved does the payload enter the ordered validation stack. The segment-level control numbers and terminators that gate this boundary for EDIFACT are validated upstream in IFCSUM EDI Message Parsing, before field extraction begins.

Python Data Structure Mapping #

Production pipelines map UN/EDIFACT segment definitions, XML customs declarations, and proprietary terminal payloads directly onto Python type systems. Declarative libraries — pydantic for regulated records, jsonschema for JSON APIs, plain TypedDict on the hot path — enforce contracts that mirror the maritime data dictionaries rather than re-inventing them; Pydantic envelope validation for UN/EDIFACT interchanges shows how the UNB/UNH envelope itself becomes a typed guard before any inner segment is trusted. The rule that makes the whole framework work is that nothing crosses the boundary as an implicit dict; every payload is deserialized into an explicitly typed model so a malformed message fails at construction, where it can be classified and routed, not three hops downstream inside a customs submission.

A typical IFCSUM instruction maps to a nested Pydantic model where each EDIFACT segment resolves to a typed field with an explicit constraint:

Coercion is deterministic. MEA+WT+G+15000:KGM becomes Decimal("15000") kilograms and stays in kilograms until the presentation boundary; a DTM value parses to a timezone-aware UTC datetime; a LOC element resolves to a five-character UN/LOCODE checked against the official registry. Weights use decimal.Decimal, never float, because a rounding error on a Verified Gross Mass (VGM) figure is a safety and compliance defect, not a display quirk. Records that must resolve into the canonical trade schema are reconciled against the Bill of Lading Schema Mapping layer, and container-level fields resolve into the Container Hierarchy Data Models topology so equipment tracking runs off a shared structure.

from __future__ import annotations

import re
from datetime import datetime
from decimal import Decimal

import structlog
from pydantic import BaseModel, Field, field_validator

log = structlog.get_logger()


class IFCSUMHeader(BaseModel):
    """Typed contract for an IFCSUM message header at the validation boundary."""

    message_ref: str = Field(..., max_length=14)
    sender_id: str
    receiver_id: str
    created_at: datetime
    gross_weight_kg: Decimal = Field(..., ge=0)

    @field_validator("message_ref")
    @classmethod
    def validate_edifact_ref(cls, v: str) -> str:
        if not re.match(r"^[A-Z0-9]{1,14}$", v):
            raise ValueError("invalid EDIFACT message reference format")
        return v

    @field_validator("created_at")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("timestamp must be timezone-aware (UTC required)")
        return v

All schema definitions are version-controlled alongside pipeline code, so backward-compatible updates ship without disrupting live vessel-reporting cycles or terminal handoff SLAs.

Validation, Quarantine & Compliance Auditing #

Validation runs in three ordered tiers inside the worker, and the tier a payload fails in decides exactly where it is routed. Classification, not the failure itself, is what keeps the line moving.

1. Structural — missing mandatory segments, invalid segment terminators, encoding faults, or payload truncation. A record that will not deserialize into its typed model raises a pydantic.ValidationError. The delimiter and envelope rules a stowage interchange must satisfy at this tier are worked through end to end in Validating BAPLIE against ISO 9735 in Python. This is transmission corruption a human cannot fix without engineering; it fails fast to the dead-letter queue.
2. Semantic — invalid UN/LOCODE values, malformed ISO 8601 timestamps, drifting ISO 6346 container check digits (weighted modulo-11), or cross-field inconsistency such as declared gross weight versus summed line items. These often stem from OCR variance or legacy terminal formatting; they are recoverable business exceptions and route to the quarantine topic, not the DLQ.
3. Regulatory — SOLAS VGM thresholds, dangerous-goods segregation rules, IMO vessel-number check digits, and HS/HTS tariff resolution against a cached registry. A gross weight exceeding net weight, a TEU count mismatching container IDs, or an HS code embargoed by the destination port is a compliance conflict that routes to review.

The distinction between the DLQ and the quarantine topic is the single most important operational decision on this page: the DLQ holds records engineering must touch; the quarantine topic holds records an ops team can reconcile against a refreshed registry without halting ingestion. The topic wiring, redrive policy, and reconciliation replay behind that split are laid out in Routing failed payloads to quarantine and a dead-letter queue. Every validation event writes an immutable audit record so a customs or port-state-control audit can replay exactly why a document was accepted, quarantined, or rejected:

{
  "correlation_id": "req_8f3a9c2d-4b11-4e9f-a8c2-1d7e9b4f0c3a",
  "timestamp": "2026-06-15T08:42:11.003Z",
  "severity": "ERROR",
  "error_class": "SEMANTIC_MISMATCH",
  "field": "LOC.port_code",
  "expected": "USNYC",
  "received": "US NYC",
  "rule_version": "v2_legacy",
  "fallback_action": "ROUTE_TO_CORRECTION_QUEUE",
  "pipeline_stage": "schema_validation"
}

Logs ship as structured JSON via structlog to a centralized observability stack (ELK, Datadog, or Prometheus/Grafana) with alert thresholds tied to error-class frequency. When structural error rates exceed 2% over a 15-minute window, the pipeline trips a circuit break and pages terminal-integration engineers, because a sudden structural-error spike almost always means a carrier changed a delimiter or an interchange format upstream.

Downstream Integration #

A payload that clears all three tiers does not become a file — it becomes a validated event other systems trust. Cleared manifests reconcile against IFCSUM EDI Message Parsing output; container-level records resolve into the Container Hierarchy Data Models so reefer monitoring and dangerous-goods segregation execute off a shared topology; lifecycle transitions propagate to the Port Call Workflow Design state machine. High-volume traffic is drained through Async Batch Processing Pipelines, which call this framework as their validation stage — the two pages own opposite halves of the same gate: batch processing owns durable delivery and back-pressure, this reference owns the rule engine that judges each payload.

Where physical movement must be correlated with position, cleared events join the real-time feed from AIS Data Stream Integration, and equipment status is reconciled through the Container Status Mapping Rules. Every cross-boundary publish authenticates under the Maritime Security Boundary Setup zero-trust controls, and the TOS reads the framework depends on for registry lookups follow the Terminal API Polling Strategies.

Fallback Chains & Uptime Guarantees #

Maritime document contracts drift, and external registries fail — the framework treats both as expected, logged modes rather than accidents. Schema versioning is the primary defense against drift: the pipeline maintains parallel validation models (v1_strict, v2_legacy, v3_relaxed) and routes payloads by detected version header. When a payload fails strict validation but passes a legacy tolerance window, it is accepted with a format_drift_warning flag and queued for batch reconciliation instead of being hard-rejected — a valid interchange is never dropped because a carrier is one revision behind.

Registry lookups are the framework’s only external dependency, and they are wrapped accordingly:

• Exponential backoff with jitter — retry transient 429/503 responses from UN/LOCODE, IMO, or customs-tariff services with a capped delay and per-attempt jitter; cap retries at three before falling back so a degraded API cannot pin the validation workers.
• Circuit breakers — open per registry endpoint after consecutive failures (500 ms timeout, 60 s recovery) and serve a cached registry snapshot instead of hammering a failing service, so ingestion throughput survives a port-authority API outage.
• Cached fallback snapshots — LOCODE, HS/HTS, and IMO tables are held locally so no worker blocks on a live customs call during a peak window; a stale snapshot degrades a payload to quarantine, never to a hard failure.
• Continuous drift telemetry — track field-level validation failure rates and regex match degradation; when a previously stable field crosses a 5% failure threshold, trigger schema regeneration and notify the carrier integration team before rejection spikes reach terminal operations.

CPU-bound pydantic validation is offloaded to a concurrent.futures.ProcessPoolExecutor so it never blocks the event loop, while I/O-bound registry lookups run concurrently via asyncio.gather(); persistent schema violations increment a failure counter and route to the dead-letter queue after three attempts.

Step-by-step Implementation Guide #

The steps below wire a versioned, tiered validation gate that any worker can call. Each step is runnable in isolation and uses type annotations with structlog for structured JSON logging.

Step 1 — Select the schema version from the payload header #

Route to the correct typed model before applying any field rules, so a legacy interchange is never judged against a newer contract.

SCHEMA_MODELS = {"D.16A": IFCSUMHeader, "D.95B": IFCSUMHeaderLegacy}


def select_model(headers: dict[str, str]) -> type[BaseModel]:
    version = headers.get("schema_version", "D.16A")
    model = SCHEMA_MODELS.get(version)
    if model is None:
        log.warning("unknown_schema_version", version=version)
        return IFCSUMHeaderLegacy  # relaxed fallback, flagged for reconciliation
    return model

Step 2 — Run the structural tier at the normalization boundary #

Deserialize into the typed model; a ValidationError here is unrecoverable corruption bound for the dead-letter queue.

from pydantic import ValidationError


def structural_tier(raw: bytes, model: type[BaseModel]) -> BaseModel | None:
    try:
        return model.model_validate_json(raw)
    except ValidationError as exc:
        log.error("structural_reject", errors=exc.errors())
        broker.publish("documents.dlq", raw, reason="STRUCTURAL")
        return None

Step 3 — Apply the semantic tier against cached registries #

Check LOCODE, ISO 6346 check digits, and cross-field consistency; recoverable failures route to quarantine, not the DLQ.

def semantic_tier(rec: IFCSUMHeader, *, locode_cache: set[str]) -> str:
    if rec.location_code not in locode_cache:
        log.warning("quarantine", field="LOC", reason="LOCODE_UNKNOWN")
        return "QUARANTINE"
    if not iso6346_check_digit_ok(rec.container_id):
        log.warning("quarantine", field="GID", reason="CHECK_DIGIT_DRIFT")
        return "QUARANTINE"
    return "PASS"

Step 4 — Apply the regulatory tier #

Enforce SOLAS VGM thresholds and tariff rules; a conflict routes to compliance review rather than acceptance.

def regulatory_tier(rec: IFCSUMHeader, *, vgm_limit_kg: Decimal) -> str:
    if rec.gross_weight_kg > vgm_limit_kg:
        log.warning("compliance_review", rule="SOLAS_VGM", value=str(rec.gross_weight_kg))
        return "COMPLIANCE_REVIEW"
    return "PASS"

Step 5 — Emit the immutable audit record and route the verdict #

Every payload, accepted or not, writes one audit record keyed on its correlation ID so the decision is replayable.

def finalize(rec: IFCSUMHeader, verdict: str, *, correlation_id: str) -> None:
    audit.write(
        correlation_id=correlation_id,
        rule_version=rec.schema_version,
        compliance_status=verdict,
        payload_sha256=rec.raw_payload_sha256,
    )
    log.info("validated", correlation_id=correlation_id, verdict=verdict)

A ValidationError in Step 2 is a structural defect for the DLQ; a QUARANTINE verdict in Step 3 is a recoverable business exception for the quarantine topic; a COMPLIANCE_REVIEW verdict in Step 4 is a regulatory conflict for a human. Keeping those three paths distinct is what lets the quarantine stream reconcile automatically once an authoritative registry recovers.

Troubleshooting Common Failures #

Frequently Asked Questions #

Related #

• Validating BAPLIE against ISO 9735 in Python — checking a stowage interchange’s envelope and delimiters against the ISO 9735 syntax rules
• Pydantic envelope validation for UN/EDIFACT interchanges — typing the UNB/UNH envelope as a guard before any inner segment is trusted
• Routing failed payloads to quarantine and a dead-letter queue — topic wiring, redrive policy, and reconciliation replay for the two failure classes
• Async Batch Processing Pipelines — the back-pressure-safe workers that call this validation gate as their validation stage
• IFCSUM EDI Message Parsing — segment-level parsing that resolves control numbers before field validation begins
• PDF Bill of Lading Extraction — OCR output promoted through this gate against carrier-specific templates
• Bill of Lading Schema Mapping — the canonical trade schema validated records resolve into
• Container Hierarchy Data Models — the container topology container-level fields validate against

Up: Document Ingestion & EDI Parsing Workflows — the parent framework governing ingestion, schema, and resilience contracts.