Skip to content

Client code

Minimal, dependency-light clients you can copy into your codebase and adapt (naming, logging, persistence). Both read the base URL and API key from EPARCEL_BASE_URL and EPARCEL_API_KEY, use a 60-second timeout, and treat 5xx and network errors as retryable with the same idempotency key.

Uses httpx.

eparcel.py
"""Minimal E-Parcel Dispatcher client."""
from __future__ import annotations


import base64
import os
import time
from typing import Any


import httpx




class EParcelError(Exception):
    """Non-2xx answer from E-Parcel. `body` is the parsed JSON (or text)."""


    def __init__(self, status: int, body: Any) -> None:
        super().__init__(f"E-Parcel answered {status}: {body}")
        self.status = status
        self.body = body


    @property
    def retryable(self) -> bool:
        # 502 = every carrier errored; other 5xx = infrastructure. Retry both
        # with the same Idempotency-Key. 4xx is final for this request.
        return self.status >= 500




class EParcelClient:
    def __init__(self, base_url: str | None = None, api_key: str | None = None,
                 timeout: float = 60.0) -> None:
        self._http = httpx.Client(
            base_url=base_url or os.environ["EPARCEL_BASE_URL"],
            headers={"Authorization": f"Bearer {api_key or os.environ['EPARCEL_API_KEY']}"},
            timeout=timeout,
        )


    def create_shipment(self, shipment: dict[str, Any], idempotency_key: str) -> dict[str, Any]:
        response = self._http.post(
            "/api/v1/shipments", json=shipment, headers={"Idempotency-Key": idempotency_key}
        )
        return _json_or_raise(response)


    def create_batch(self, shipments: list[dict[str, Any]]) -> list[dict[str, Any]]:
        response = self._http.post("/api/v1/shipments/batch", json={"shipments": shipments})
        return _json_or_raise(response)["shipments"]


    def get_shipment(self, shipment_id: int) -> dict[str, Any]:
        return _json_or_raise(self._http.get(f"/api/v1/shipments/{shipment_id}"))


    def get_label(self, shipment_id: int) -> bytes:
        response = self._http.get(f"/api/v1/shipments/{shipment_id}/label")
        if response.status_code != 200:
            raise EParcelError(response.status_code, _body(response))
        return response.content




def ship_parcel(client: EParcelClient, shipment: dict[str, Any], idempotency_key: str,
                delays: tuple[float, ...] = (5, 30, 120)) -> dict[str, Any]:
    """Create one shipment, retrying unknown outcomes and 5xx with the same key."""
    for delay in (*delays, None):
        try:
            result = client.create_shipment(shipment, idempotency_key)
            result["label_pdf"] = base64.b64decode(result.pop("label_pdf_base64"))
            return result
        except httpx.TransportError:
            if delay is None:  # timeout or connection error: outcome unknown
                raise
        except EParcelError as exc:
            if not exc.retryable or delay is None:
                raise
        time.sleep(delay)
    raise AssertionError("unreachable")




def _json_or_raise(response: httpx.Response) -> Any:
    if response.status_code == 200:
        return response.json()
    raise EParcelError(response.status_code, _body(response))




def _body(response: httpx.Response) -> Any:
    try:
        return response.json()
    except ValueError:
        return response.text

In a web request or job worker, enqueue the retries rather than sleeping inline.

Generating types from the schema

Section titled “Generating types from the schema”

The OpenAPI document is published by every environment at /openapi.json. To generate TypeScript types from it instead of writing them by hand:

Terminal window
npx openapi-typescript https://api-test.e-parcel.fi/openapi.json -o eparcel.d.ts