Skip to content

Integrations

Database-side primitives for message-driven consumers (§28) and micro-batching (§17.1). dbkit is scoped as database-only and does not ship a broker-facing adapter — an application wires these into its own consumer loop.

inbox_ddl 

inbox_ddl(table: str = DEFAULT_INBOX_TABLE) -> str

DDL for the inbox table (§28.3). Time-partition it in production for cheap pruning — see :func:partitioned_inbox_ddl/:func:inbox_month_partition_ddl for that variant.

partitioned_inbox_ddl 

partitioned_inbox_ddl(
    table: str = DEFAULT_INBOX_TABLE,
) -> str

DDL for a partitioned inbox table, range-partitioned by processed_at (§28.3).

A single unbounded consumed_messages table's dedup lookup gets slower and its storage grows without bound as message volume accumulates. Range-partitioning by month makes pruning old data a cheap DROP TABLE on a partition instead of a slow DELETE. The partition key must be part of the primary key for declarative partitioning, so this variant keys on (consumer_name, message_id, processed_at) instead of just the first two.

Partitions themselves aren't created automatically — call :func:inbox_month_partition_ddl for each month you need (typically from a scheduled job that creates the next 1-2 months ahead of time; pg_partman or a cron-triggered dbkit script both work). Older partitions can simply be dropped once no longer needed for dedup/audit purposes — a message's redelivery window is bounded by the broker's own retention, so a partition older than that window is always safe to drop.

inbox_month_partition_ddl 

inbox_month_partition_ddl(
    year: int,
    month: int,
    *,
    table: str = DEFAULT_INBOX_TABLE,
) -> str

DDL for one calendar-month partition of :func:partitioned_inbox_ddl's table.

year/month are passed explicitly (not computed from the current date) so this stays a pure function — call it with whatever month you actually need created, e.g. from a scheduled job that creates the next month a few days before it starts.

process_once async 

process_once(
    db: Any,
    *,
    consumer: str,
    message_id: str,
    target: DatabaseTarget,
    inbox_table: str = DEFAULT_INBOX_TABLE,
) -> AsyncIterator[tuple[Any, bool]]

Open a transaction, claim the message in the inbox, and yield (tx, first_time).

Do the business writes on tx only when first_time is True; a duplicate delivery still commits (a harmless no-op) so the broker message can be acked::

async with process_once(db, consumer="c", message_id=mid, target=t) as (tx, first):
    if first:
        await tx.execute(INSERT_ORDER, params)

ack_after_commit async 

ack_after_commit(
    db: Any,
    *,
    consumer: str,
    message_id: str,
    target: DatabaseTarget,
    work: Callable[[Any], Awaitable[None]],
    ack: Callable[[], Awaitable[None]],
    dead_letter: Callable[[DatabaseError], Awaitable[None]]
    | None = None,
    retry: Callable[[DatabaseError], Awaitable[None]]
    | None = None,
    inbox_table: str = DEFAULT_INBOX_TABLE,
) -> bool

Run work idempotently, then ack — the full §28 consume flow.

Returns True if the message was processed (or was a duplicate) and acked. Behavior on failure:

  • commit-unknown -> re-raised; the message is not acked (idempotent replay will dedupe).
  • retryable error -> retry callback if given, else re-raised (nack/requeue).
  • permanent error -> dead_letter callback if given, else re-raised (route to DLQ).

Transactional outbox (§28.4)

The mirror of the inbox: enqueue an event in the same transaction as the business write so the event exists iff the change persisted, then relay unsent rows to a broker at-least-once. Single-shard (the outbox table lives on the same database as the write); pair with the inbox on the consumer for effectively-once. Schema-agnostic — you pass the table name and an opaque JSON payload.

from dbkit.integrations import outbox_ddl, enqueue, drain

await db.execute(sql(outbox_ddl()), target=t)   # once, at setup (script: split on ';')

async with db.transaction(target=t) as tx:       # atomic: write + event
    await tx.execute(UPDATE_ORDER, params)
    await enqueue(tx, topic="order.completed", payload={"order_id": oid})

# relay loop (separate worker): publish unsent rows, mark them sent
await drain(db, target=t, publish=my_async_publish)

outbox_ddl 

outbox_ddl(table: str = DEFAULT_OUTBOX_TABLE) -> str

DDL for the outbox table (§28.4).

Columns: id (monotonic bigserial — relay order + cursor), topic (routing key), payload (opaque jsonb event body), created_at, and sent_at (NULL until the relay publishes it). A partial index on the unsent rows keeps the relay's "next batch" scan cheap regardless of how many already-sent rows remain. Time-partition it in production for cheap pruning of sent rows — see :func:partitioned_outbox_ddl/:func:outbox_month_partition_ddl.

partitioned_outbox_ddl 

partitioned_outbox_ddl(
    table: str = DEFAULT_OUTBOX_TABLE,
) -> str

DDL for a partitioned outbox table, range-partitioned by created_at (§28.4).

An unbounded outbox grows without limit as events accumulate; range-partitioning by month makes pruning old (already-sent) data a cheap DROP TABLE on a partition instead of a slow DELETE. The partition key must be part of the primary key for declarative partitioning, so this variant keys on (id, created_at). Create each month's partition with :func:outbox_month_partition_ddl (typically from a scheduled job a month ahead); drop partitions once every row in them has been sent and is past any audit window.

outbox_month_partition_ddl 

outbox_month_partition_ddl(
    year: int,
    month: int,
    *,
    table: str = DEFAULT_OUTBOX_TABLE,
) -> str

DDL for one calendar-month partition of :func:partitioned_outbox_ddl's table.

year/month are passed explicitly (not computed from the current date) so this stays a pure function — call it with whatever month you need created, e.g. from a scheduled job that creates the next month a few days before it starts.

enqueue async 

enqueue(
    tx: Any,
    *,
    topic: str,
    payload: Mapping[str, Any],
    table: str = DEFAULT_OUTBOX_TABLE,
) -> int

Insert one event into the outbox on the caller's transaction tx and return its id.

Call this inside a db.transaction(...) block alongside the business writes so the event and the change commit atomically::

async with db.transaction(target=t) as tx:
    await tx.execute(UPDATE_ORDER, params)
    await enqueue(tx, topic="order.completed", payload={"order_id": oid})

payload is serialized to jsonb and bound as a parameter (never interpolated).

drain async 

drain(
    db: Any,
    *,
    target: DatabaseTarget,
    publish: Callable[[Mapping[str, Any]], Awaitable[None]],
    batch_size: int = 100,
    table: str = DEFAULT_OUTBOX_TABLE,
) -> int

Relay a batch of unsent outbox rows to publish, then mark them sent. Returns the count.

Claims up to batch_size unsent rows in id order with FOR UPDATE SKIP LOCKED — so many relay workers can run concurrently without blocking or double-claiming — publishes each via the publish callback (receives a mapping with id/topic/payload/created_at), then stamps sent_at for the whole batch and commits. Everything runs in one transaction: if publish raises, the claim rolls back and the rows are retried on the next :func:drain.

Delivery is at-least-once: a crash after the broker accepts a message but before commit replays it. Make publish idempotent or dedupe on the consumer (:mod:dbkit.integrations.inbox). Loop :func:drain from a relay worker (a return of 0 means the outbox is momentarily empty).

BatchCollector 

BatchCollector(
    flush: Callable[[Sequence[T]], Awaitable[None]],
    *,
    max_size: int = 1000,
    max_delay_ms: float = 50.0,
)

Bases: Generic[T]

Aggregates items across concurrent producers and flushes them as one batch (§17.1).

flush is called with the buffered batch whenever max_size/max_delay_ms is reached; it should perform the batched write (e.g. db.insert_many).

add async 

add(item: T) -> None

Add an item; flush immediately if the buffer is full, else arm the flush timer.

flush async 

flush() -> None

Flush any buffered items now.

close async 

close() -> None

Flush remaining items and stop the timer.

HTTP APIs (FastAPI / Starlette)

Category-driven RFC 7807 responses for unhandled DatabaseErrors: overload (pool timeout, limiter, circuit open, backend unreachable) → 503 with a Retry-After header (derived from the circuit breaker's open_seconds when you pass your facade); query timeouts → 504; programming/integrity errors stay 500 and loud. Requires the dbkit-sql[fastapi] extra (Starlette only).

from dbkit.integrations.fastapi import install_exception_handlers

install_exception_handlers(app, database=db)

install_exception_handlers 

install_exception_handlers(
    app: Starlette | Any,
    *,
    retry_after_seconds: int | None = None,
    database: Any = None,
    expose_detail: bool = False,
) -> None

Register a :class:DatabaseError handler on app (FastAPI or Starlette).

retry_after_seconds populates the 503 Retry-After header. When omitted and database (an AsyncDatabase/Database) is given, it derives from the circuit breaker's open_seconds — the earliest moment a retry can find the breaker half-open. Falls back to 2 seconds.

expose_detail includes the sanitized error message in the response body — keep it off in production unless the API is internal.

Per-query session settings

Query.settings applies transaction-local PostgreSQL settings right before the statement — parameterized set_config(name, value, true), names validated:

COUNTS = Query(
    name="inbox.counts",
    statement=sql("SELECT ..."),
    settings={"jit": "off"},          # or {"work_mem": "64MB"} for a heavy aggregate
)

The setting never leaks: it resets when the statement's transaction ends, before the connection returns to the pool.