FastAPI adapter for the python-sendparcel shipping ecosystem.

Alpha notice — This package is at version 0.1.1 and its API is not yet stable. Breaking changes may occur in minor releases until 1.0.

• Router factory — single call to create_shipping_router() gives you a fully-configured APIRouter with shipment, label, status and callback endpoints.
• Provider-agnostic — plug in any shipping provider that implements the python-sendparcel provider protocol.
• Plugin registry — FastAPIPluginRegistry discovers and manages provider plugins with optional per-provider routers.
• Pydantic-native configuration — SendparcelConfig reads from environment variables with the SENDPARCEL_ prefix.
• Webhook callback handling — built-in endpoint for provider status callbacks with automatic retry queue support.
• SQLAlchemy contrib — optional [sqlalchemy] extra provides SQLAlchemyShipmentRepository, SQLAlchemyRetryStore, and ready-made database models.
• Exception mapping — core sendparcel exceptions are automatically converted to appropriate HTTP status codes (400, 404, 409, 502).
• Async-first — fully asynchronous with async/await throughout.

Install the base package:

pip install fastapi-sendparcel

If you want SQLAlchemy-backed persistence (recommended):

pip install fastapi-sendparcel[sqlalchemy]

Note: The project uses uv for development. If you are contributing, run uv sync instead.

Below is a minimal but complete FastAPI application that wires up the shipping router with SQLAlchemy persistence.

from contextlib import asynccontextmanager
from collections.abc import AsyncGenerator

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from fastapi_sendparcel import (
    SendparcelConfig,
    FastAPIPluginRegistry,
    create_shipping_router,
)
from fastapi_sendparcel.contrib.sqlalchemy.models import Base
from fastapi_sendparcel.contrib.sqlalchemy.repository import (
    SQLAlchemyShipmentRepository,
)
from fastapi_sendparcel.contrib.sqlalchemy.retry_store import (
    SQLAlchemyRetryStore,
)

# --- Database ---
engine = create_async_engine("sqlite+aiosqlite:///./shipments.db")
async_session = async_sessionmaker(engine, class_=AsyncSession)

# --- Sendparcel setup ---
config = SendparcelConfig(
    default_provider="my-provider",
    providers={
        "my-provider": {
            "api_key": "...",
        },
    },
)

repository = SQLAlchemyShipmentRepository(async_session)
retry_store = SQLAlchemyRetryStore(async_session)
registry = FastAPIPluginRegistry()

shipping_router = create_shipping_router(
    config=config,
    repository=repository,
    registry=registry,
    retry_store=retry_store,
)

# --- App ---
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()


app = FastAPI(title="My Shipping App", lifespan=lifespan)
app.include_router(shipping_router, prefix="/api/shipping")

The create_shipping_router function accepts all its arguments as keyword-only parameters:

SendparcelConfig extends Pydantic's BaseSettings and reads environment variables with the SENDPARCEL_ prefix.

You can instantiate the config directly or let it read from the environment:

# Explicit values
config = SendparcelConfig(
    default_provider="inpost",
    providers={"inpost": {"api_key": "secret"}},
)

# From environment variables
# (set SENDPARCEL_DEFAULT_PROVIDER=inpost, etc.)
config = SendparcelConfig()

The router created by create_shipping_router() exposes the following endpoints. All paths are relative to the prefix you mount the router at (e.g. /api/shipping).

POST /shipments — request body:

{
  "reference_id": "my-ref-123",
  "provider": "my-provider",
  "sender_address": {
    "name": "John Smith",
    "line1": "1 Example St",
    "city": "Warsaw",
    "postal_code": "00-001",
    "country_code": "PL"
  },
  "receiver_address": {
    "name": "Jane Doe",
    "line1": "5 Destination St",
    "city": "Krakow",
    "postal_code": "30-001",
    "country_code": "PL"
  },
  "parcels": [
    {"weight_kg": 2.5}
  ]
}

The provider field is optional; when omitted, default_provider from the config is used. The reference_id field is optional and can be used for external reference tracking.

ShipmentOperationResponse — returned by shipment, label and status endpoints:

{
  "id": "abc-def",
  "status": "label_ready",
  "provider": "my-provider",
  "external_id": "EXT123",
  "tracking_number": "TRACK456",
  "label": {
    "format": "PDF",
    "url": "https://...",
    "content_base64": null
  },
  "update": null
}

Label payloads are returned as operation results. They are not persisted on the shipment model.

CallbackResponse — returned by the callback endpoint:

{
  "provider": "my-provider",
  "status": "accepted",
  "shipment": {
    "id": "abc-def",
    "status": "in_transit",
    "provider": "my-provider",
    "external_id": "EXT123",
    "tracking_number": "TRACK456"
  },
  "update": {
    "status": "in_transit",
    "tracking_number": "TRACK456",
    "tracking_events": []
  }
}

The router automatically registers exception handlers that map core sendparcel exceptions to HTTP responses:

Stores failed webhook callbacks for retry processing. The SQLAlchemy contrib provides a ready-made implementation (SQLAlchemyRetryStore), but you can implement this protocol with any backend (Redis, DynamoDB, etc.).

from fastapi_sendparcel import CallbackRetryStore


class MyRetryStore:
    async def store_failed_callback(
        self, shipment_id: str, payload: dict, headers: dict
    ) -> str: ...

    async def get_due_retries(self, limit: int = 10) -> list[dict]: ...

    async def mark_succeeded(self, retry_id: str) -> None: ...

    async def mark_failed(self, retry_id: str, error: str) -> None: ...

    async def mark_exhausted(self, retry_id: str) -> None: ...

The optional [sqlalchemy] extra provides production-ready persistence components:

• ShipmentModel — SQLAlchemy model mapped to the sendparcel_shipments table.
• CallbackRetryModel — SQLAlchemy model mapped to the sendparcel_callback_retries table.
• SQLAlchemyShipmentRepository — async repository implementing the ShipmentRepository protocol.
• SQLAlchemyRetryStore — async retry store implementing the CallbackRetryStore protocol.

Both require an async_sessionmaker[AsyncSession] at construction time:

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from fastapi_sendparcel.contrib.sqlalchemy.repository import (
    SQLAlchemyShipmentRepository,
)
from fastapi_sendparcel.contrib.sqlalchemy.retry_store import (
    SQLAlchemyRetryStore,
)

session_factory = async_sessionmaker(engine, class_=AsyncSession)

repository = SQLAlchemyShipmentRepository(session_factory)
retry_store = SQLAlchemyRetryStore(session_factory, backoff_seconds=60)

The example/ directory contains a full demo application with:

• Tabler-based UI with Jinja2 templates and HTMX
• Shipment creation with sender/receiver address forms
• Label generation and PDF download
• A simulated delivery provider (delivery_sim.py)

cd example
uv sync
uv run uvicorn app:app --reload

Then open http://localhost:8000 in your browser.

uv sync --all-extras
uv run pytest

The test suite uses pytest with pytest-asyncio in auto mode. Configuration is in pyproject.toml.

Created and maintained by Dominik Kozaczko.

This project is the FastAPI adapter for the python-sendparcel ecosystem.

MIT