Outbox and relay¶

The transactional outbox pattern solves the dual-write problem: a domain event is written to the database in the same transaction as the business row, so the event cannot be lost even if the broker is temporarily unavailable. A background worker (Relay) then drains the outbox table and delivers events to the broker with at-least-once semantics.

pip install "alchemiq[outbox,postgres]"

The outbox table¶

Import of alchemiq.outbox creates the outbox table automatically. OutboxEvent is the ORM model for each row:

Column	Type	Description
`id`	int PK	Auto-assigned surrogate key
`topic`	str	Broker routing key, e.g. `"user.signed_up"`
`aggregate_type`	str (nullable)	Optional model name
`aggregate_id`	str (nullable)	Optional partition key
`event_type`	str (nullable)	Optional sub-type discriminator
`payload`	JSON	Event data
`headers`	JSON (nullable)	Optional broker-level headers
`status`	str	`pending` -> `published` / `failed` -> `dead`
`attempts`	int	Delivery attempt counter
`created_at`	timestamptz	Row creation time
`published_at`	timestamptz (nullable)	Successful delivery timestamp
`last_error`	str (nullable)	Last delivery exception message

Publishing events manually¶

publish() writes an outbox row in its own autocommit transaction. Use it for manual event emission from background tasks or CLI scripts. To tie the event to a business transaction, wrap the call in a UnitOfWork.

from alchemiq import publish

# simple dict payload
await publish("user.signed_up", {"id": 1, "email": "ada@example.com"})

# Pydantic payload - dumped automatically
await publish("billing.upgraded", BillingUpgraded(plan="pro"), key="user-42")

Parameters:

Parameter	Description
`topic`	Broker routing key
`payload`	`dict` or Pydantic `BaseModel`
`key`	Optional partition key (stored as `aggregate_id`)
`headers`	Optional broker-level headers dict

Automatic capture via model signals¶

If a model has Meta.outbox = True, alchemiq writes an outbox row automatically on every create, update, or delete - in the same transaction as the business operation. No manual publish call is required.

from alchemiq import Model
from alchemiq.types import PK

class Order(Model):
    id: PK[int]
    status: str

    class Meta:
        outbox = True

The Publisher protocol¶

Publisher is a structural (duck-typed) protocol. Any object that exposes an async publish(message) method satisfies it:

from alchemiq import OutboxMessage
from alchemiq.outbox import Publisher, TransientPublishError

class MyBrokerPublisher:
    async def publish(self, message: OutboxMessage) -> None:
        try:
            await broker.send(message.topic, message.payload)
        except BrokerConnectionError as e:
            raise TransientPublishError(str(e)) from e

Raise TransientPublishError for connection failures - the relay backs off without incrementing attempts. Any other exception poisons the row (increments attempts, marks it failed or dead).

An optional publish_batch(messages) method is detected by duck-typing when present. It is not part of the protocol contract, so any object with only publish satisfies isinstance(obj, Publisher).

The Relay worker¶

Relay drains the outbox table and delivers rows to a broker. Run it as an asyncio background task:

import asyncio
from alchemiq.outbox import Relay

relay = Relay(my_publisher, batch_size=50, poll_interval=2.0, max_attempts=5)

task = asyncio.create_task(relay.run())

# on shutdown:
relay.stop()
await task

What Relay claims each cycle¶

Each cycle claims a batch of both pending and failed rows with FOR UPDATE SKIP LOCKED. This makes it safe to run multiple concurrent workers without double-delivery.

Error taxonomy¶

Error type	Behaviour
`TransientPublishError` (per-message or batch)	Whole-batch rollback; `attempts` is not incremented; relay sleeps `error_backoff` seconds
Any other exception - per-message path	Row poisoned: `attempts` incremented; status -> `failed` or `dead`
Any other exception - `publish_batch` path	All rows in the batch are poisoned together (not row-by-row)

The publish_batch poison behaviour means a single bad batch poisons the entire set of claimed rows. Use per-message publish if you need finer fault isolation.

Parameters¶

Parameter	Default	Description
`publisher`	(required)	Delivery adapter satisfying the `Publisher` protocol
`batch_size`	100	Maximum rows claimed per cycle
`poll_interval`	1.0 s	Wait between cycles when the batch was not full
`max_attempts`	5	Attempts before a row is marked `dead`
`error_backoff`	5.0 s	Sleep after a transient broker error

FastStream publishing¶

The [faststream] extra provides FastStreamPublisher, a Publisher adapter that works with any FastStream broker (RabbitMQ, Kafka, NATS, Redis):

pip install "alchemiq[faststream,postgres]"

from faststream.rabbit import RabbitBroker
from alchemiq import Relay
from alchemiq.faststream import FastStreamPublisher

broker = RabbitBroker("amqp://guest:guest@localhost/")
await broker.connect()

relay = Relay(FastStreamPublisher(broker), batch_size=200)
await relay.run()

The correlation_id of each published message is set to str(message.id) (the outbox row PK) so consumers can deduplicate under the at-least-once contract. Event metadata (aggregate_type, aggregate_id, event_type) is forwarded as broker headers under the alchemiq.* prefix.

FastStreamPublisher exposes only publish (no publish_batch), so the relay always uses the per-message delivery path with it - each row is published and acknowledged individually rather than as a batch.

Consumer dependency injection¶

alchemiq.faststream exposes the same providers as alchemiq.fastapi. Use them as Depends targets in FastStream subscribers to inject a session, unit of work, or repository:

from faststream import Depends, FastStream
from alchemiq.faststream import lifespan, unit_of_work
from alchemiq import Repository

app = FastStream(broker, lifespan=lifespan("postgresql+asyncpg://user:pass@localhost/mydb"))

@broker.subscriber("orders.created")
async def on_order(evt: dict, uow=Depends(unit_of_work)):
    async with uow:
        order = await Repository(Order).create(**evt)

Outbox capture fires automatically when the subscriber writes through the unit of work.