Polling terminal operating systems via REST APIs #

This guide shows how to construct the actual REST calls that pull container status, gate transactions, and equipment moves from a Terminal Operating System (TOS) — the headers, authentication, pagination cursors, conditional requests, and payload parsing that turn a raw endpoint into a rate-safe, resumable event source.

Architecture Alignment #

Request construction is the lowest layer of the Terminal API Polling Strategies topic: the parent page defines when to poll (adaptive cadence per vessel phase) and how the loop degrades (the tiered fallback chain), while this page defines what one request looks like on the wire. Both sit inside the Container Tracking & AIS Event Synchronization domain, whose job is to fuse landside TOS pulls with the push-based AIS Data Stream Integration feed into one trustworthy container state. Get the request wrong — a hard-coded delimiter, a full re-fetch every cycle, an unhandled 429 — and every downstream consumer inherits stale or throttled data. The typed events this layer emits are resolved by the Container Status Mapping Rules engine before any state is committed.

Prerequisites & Environment Setup #

• Python 3.11+ for datetime.fromisoformat full ISO 8601 support and tomllib.
• Packages: httpx (async HTTP/2 client), pydantic>=2 (typed validation), structlog (structured JSON logs), tenacity (retry/backoff), orjson (fast, strict JSON), and redis (cursor persistence).
• Credentials: a per-facility OAuth2 client credential or bearer token, scoped to the terminal zone that owns the events. Never share one credential across facilities — the audit trail must attribute every pull to a single source.
• Environment variables: TOS_BASE_URL, TOS_CLIENT_ID, TOS_CLIENT_SECRET, TOS_FACILITY_LOCODE (the UN/LOCODE of the terminal), and REDIS_URL for the cursor store.

python -m pip install "httpx[http2]" "pydantic>=2" structlog tenacity orjson redis
export TOS_BASE_URL="https://n4.exampleport.com/apex/v1"
export TOS_FACILITY_LOCODE="NLRTM"

TOS vendors — Navis N4, Tideworks Mainsail, Kaleris, and the port community systems that wrap them — expose events over RESTful JSON, so the same client shape works across all of them once the base URL and auth flow are configured. Container identity is always resolved back against the Container Hierarchy Data Models so a bare box number carries its size-type and grouping.

Step-by-step Implementation #

Each step is runnable in isolation and uses type annotations with structlog — bare print() is never acceptable in a pipeline that has to be audited.

Step 1 — Define a drift-tolerant typed event model #

TOS vendors iterate endpoints without strict backward compatibility: a gate_in_timestamp may shift from an ISO 8601 string to a Unix epoch integer, or nested container_events may flatten into top-level keys during a platform upgrade. Bind the payload to a pydantic model with extra="ignore" so vendor-added noise is discarded, and normalise the timestamp regardless of its wire form. The ISO 6346 container reference is validated here per the ISO 6346 owner-prefix + serial + check-digit structure.

import structlog
from datetime import datetime, timezone
from typing import Optional
from pydantic import BaseModel, ConfigDict, field_validator

log = structlog.get_logger()


class TOSContainerEvent(BaseModel):
    model_config = ConfigDict(extra="ignore")

    container_id: str
    event_code: str
    timestamp_utc: datetime
    location: Optional[str] = None
    raw_payload: dict  # immutable audit trail

    @field_validator("timestamp_utc", mode="before")
    @classmethod
    def normalize_timestamp(cls, v: object) -> datetime:
        if isinstance(v, (int, float)):
            return datetime.fromtimestamp(v, tz=timezone.utc)
        if isinstance(v, str):
            return datetime.fromisoformat(v.replace("Z", "+00:00"))
        raise ValueError(f"Unsupported timestamp format: {type(v)}")

Step 2 — Build one long-lived async client with auth and explicit timeouts #

Amortise the TCP and TLS handshake across cycles with a persistent connection pool; a single facility integration fans out to dozens of endpoints, and per-request handshakes dominate latency otherwise. Send Accept: application/json and a bearer credential on every call, and set a hard read timeout so a stalled endpoint cannot hang the event loop.

import httpx

def build_client(base_url: str, token: str) -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url=base_url,
        headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
        limits=httpx.Limits(max_keepalive_connections=20, max_connections=100),
        timeout=httpx.Timeout(connect=5.0, read=15.0, write=5.0, pool=5.0),
        http2=True,
    )

Step 3 — Fetch one page with a conditional, cursor-based request #

Never re-fetch the whole dataset. Carry a monotonic cursor (the terminal’s sequence_id) so a mid-cycle failure resumes from the last acknowledged token, and send the stored ETag in If-None-Match so an unchanged resource returns 304 Not Modified — zero payload transferred, zero cost against the rate budget.

async def fetch_page(
    client: httpx.AsyncClient, endpoint: str, cursor: str, etag: str | None
) -> httpx.Response:
    headers = {"If-None-Match": etag} if etag else {}
    resp = await client.get(endpoint, params={"cursor": cursor, "limit": 500}, headers=headers)
    log.info("poll.fetch", endpoint=endpoint, cursor=cursor, status=resp.status_code)
    return resp

Step 4 — Parse, honouring 304 and reading the body once #

A JSON document cannot be decoded from arbitrary, non-aligned network chunks, so read the full body then parse once with orjson. Treat 304 as “no change, keep the cursor,” and route 429 to the backoff handler with any Retry-After the TOS supplied.

import orjson

def parse_body(resp: httpx.Response) -> list[dict]:
    if resp.status_code == 304:
        return []
    resp.raise_for_status()
    payload = orjson.loads(resp.content)
    # TOS feeds vary: some wrap rows in {"events": [...]}, others return a bare list
    return payload["events"] if isinstance(payload, dict) else payload

Step 5 — Normalise each row into a typed event with a payload hash #

Hash the canonical (sorted-key) payload so identical rows hash identically — this is the idempotency key that lets retried pages dedupe, and the audit fingerprint required for SOLAS/ISPS review. The same append-only discipline governs the Bill of Lading Schema Mapping audit trail.

import hashlib

def to_event(row: dict) -> TOSContainerEvent:
    fingerprint = hashlib.sha256(
        orjson.dumps(row, option=orjson.OPT_SORT_KEYS)
    ).hexdigest()
    return TOSContainerEvent(
        container_id=row["container_id"],
        event_code=row["event_code"],
        timestamp_utc=row["timestamp_utc"],
        location=row.get("yard_block"),
        raw_payload={**row, "_sha256": fingerprint},
    )

Step 6 — Wrap the request in retry, backoff, and Retry-After handling #

Retry only recoverable status codes (429, 502, 503) with exponential backoff and jitter to avoid a thundering herd when a terminal comes back online; never retry 400/401/404. When the TOS sends a delta-seconds Retry-After, honour it exactly; when it sends an HTTP-date or nothing, fall through to computed backoff.

from tenacity import retry, stop_after_attempt, wait_exponential_jitter

RECOVERABLE = {429, 502, 503}


def retry_after_seconds(resp: httpx.Response) -> float | None:
    value = resp.headers.get("Retry-After")
    if value is None:
        return None
    try:
        return float(value)          # delta-seconds form
    except ValueError:
        return None                  # HTTP-date form → let backoff decide


@retry(stop=stop_after_attempt(5),
       wait=wait_exponential_jitter(initial=2, max=60, jitter=3))
async def poll_once(client: httpx.AsyncClient, endpoint: str, cursor: str,
                    etag: str | None) -> httpx.Response:
    resp = await fetch_page(client, endpoint, cursor, etag)
    if resp.status_code in RECOVERABLE:
        log.warning("poll.recoverable", status=resp.status_code,
                    retry_after=retry_after_seconds(resp), endpoint=endpoint)
        resp.raise_for_status()
    return resp

Step 7 — Persist the cursor and ETag, then publish #

Advance the cursor and store the response ETag atomically only after the events are acknowledged on the bus, so a crash mid-publish replays rather than skips. Prune tokens older than 24 hours to bound Redis growth on long-running workers. Resolved states go on to feed the Port Call Workflow Design state machine.

async def commit(redis, endpoint: str, resp: httpx.Response,
                 events: list[TOSContainerEvent], bus) -> None:
    for ev in events:
        await bus.publish("terminal.events", ev)
    if events:
        await redis.set(f"cursor:{endpoint}", events[-1].raw_payload["sequence_id"])
    if etag := resp.headers.get("ETag"):
        await redis.set(f"etag:{endpoint}", etag)
    log.info("poll.commit", endpoint=endpoint, published=len(events),
             fallback_activated=False)

Edge Cases & Carrier Deviations #

• Retry-After as an HTTP-date. RFC 9110 permits either delta-seconds or an HTTP-date. Parsing only the integer form and silently discarding the date form (Step 6) means the client falls back to computed backoff instead of crashing — but if a TOS routinely sends dates, parse them so you honour the vendor’s real window.
• Weak vs strong ETags. Some port community systems emit weak validators (W/"abc"). They are still valid for If-None-Match and still return 304; do not strip the W/ prefix or you will force full re-fetches.
• Cursor resets on vendor upgrade. After a major TOS release, sequence_id numbering can restart. Detect a cursor that suddenly points past the newest row (an empty page with a lower max sequence than stored) and re-baseline from the earliest available token rather than looping forever on 304.
• UN/LOCODE with 0/1 in the location code. The location portion uses letters A–Z and digits 2–9; 0 and 1 are excluded to avoid O/I confusion. A payload carrying RT0 for Rotterdam is a vendor bug — upper-case, then route unresolved codes to quarantine, never to the state store.
• Duplicate gate moves on retry. A redelivered page must not double-count. Keying on the _sha256 fingerprint from Step 5 makes repeat rows no-ops.
• Unbounded memory on long runs. Accumulating full-payload lists in the event loop leaks. Stream-parse large feeds, yield events from generators, reuse the one AsyncClient, and evict acknowledged tokens after 24 hours.

Verification & Testing #

Assert three things: schema drift is absorbed, a 304 yields no events, and a 429 triggers backoff rather than a crash. The fixture below feeds an epoch-integer timestamp (the drifted form) and confirms it normalises to a timezone-aware UTC datetime.

import pytest

RAW_ROW = {
    "container_id": "MSKU1234567",
    "event_code": "GATE_IN",
    "timestamp_utc": 1719964800,      # epoch int — drifted from ISO 8601 string
    "sequence_id": 42,
    "yard_block": "A12",
}


def test_epoch_timestamp_normalises_to_utc():
    ev = to_event(RAW_ROW)
    assert ev.timestamp_utc.tzinfo is not None
    assert ev.timestamp_utc.year == 2024
    assert ev.raw_payload["_sha256"]  # audit fingerprint present


def test_304_yields_no_events():
    resp = httpx.Response(status_code=304, request=httpx.Request("GET", "/events"))
    assert parse_body(resp) == []


def test_vendor_noise_is_ignored():
    row = {**RAW_ROW, "unexpected_vendor_field": "ignore-me"}
    ev = to_event(row)
    assert not hasattr(ev, "unexpected_vendor_field")

Expected structured-log output for a healthy cycle followed by a throttled one — each line is one JSON object your log store can index on status and retry_after:

{"event": "poll.fetch", "endpoint": "/events", "cursor": "42", "status": 200}
{"event": "poll.commit", "endpoint": "/events", "published": 3, "fallback_activated": false}
{"event": "poll.fetch", "endpoint": "/events", "cursor": "45", "status": 429}
{"event": "poll.recoverable", "status": 429, "retry_after": 12.0, "endpoint": "/events"}

A missing poll.commit between two poll.fetch lines is itself an alerting signal — wire it into the Threshold Tuning for Alerts layer so a silently stalled endpoint pages an operator.

Frequently Asked Questions #

Related #

• Terminal API Polling Strategies — the adaptive cadence and tiered fallback chain this request layer plugs into.
• Container Status Mapping Rules — deterministic state resolution the polled events feed into.
• AIS Data Stream Integration — the push-based vessel feed cross-validated against terminal-reported moves.
• Threshold Tuning for Alerts — turning cycle telemetry into transient-vs-systemic alerting.
• Container Hierarchy Data Models — the equipment topology every polled box number resolves against.

↑ Back to Terminal API Polling Strategies.