Outbox and relay

class alchemiq.OutboxEvent(**kwargs)

Bases: Model

Persistent outbox row storing a domain event awaiting broker delivery.

Status lifecycle: pending -> published (success) or failed (delivery error, retryable) -> dead (max_attempts reached).

See also

Relay - background worker that drains this table.

class Meta

Bases: object

Maps to the outbox table with a (status, id) index for efficient polling.

class alchemiq.OutboxMessage(id, topic, payload, headers, aggregate_type, aggregate_id, event_type)

Bases: object

Immutable, broker-agnostic view of one outbox row handed to a Publisher.

Produced from an OutboxEvent row by the Relay before each delivery attempt. Consumers should deduplicate on id.

Parameters:

• id (int)

• topic (str)

• payload (dict[str, Any])

• headers (dict[str, Any] | None)

• aggregate_type (str | None)

• aggregate_id (str | None)

• event_type (str | None)

class alchemiq.Publisher(*args, **kwargs)

Bases: Protocol

Structural protocol for outbox delivery adapters.

Implementations must expose publish(message). An optional publish_batch(messages) method is detected by duck-typing when present; it is not part of this protocol contract, so any object with just publish satisfies isinstance(obj, Publisher).

E.g.:

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

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

See also

Relay - background loop that calls this protocol.

async publish(message)

Deliver a single message to the broker.

Raise TransientPublishError for connection failures; the relay backs off without incrementing attempts. Raise any other exception to poison the row (increment attempts; mark failed or dead).

Parameters:

message (OutboxMessage) – the outbox row projected to a broker-agnostic value object.

Raises:

TransientPublishError – for transient broker connection failures.

Return type:

None

class alchemiq.Relay(publisher, *, batch_size=100, poll_interval=1.0, max_attempts=5, error_backoff=5.0)

Bases: object

Background relay that drains the outbox table and delivers rows to a broker.

Each cycle claims a batch with FOR UPDATE SKIP LOCKED (safe for multiple concurrent relay workers), publishes each row, and records the outcome within the same transaction. Both pending and failed rows are claimed and re-attempted every cycle until they reach dead.

Delivery semantics: at-least-once - consumers must deduplicate on OutboxMessage.id.

Error taxonomy:

• TransientPublishError -> whole-batch rollback; attempts is not incremented; relay sleeps error_backoff seconds before the next cycle.

• Any other exception (per-message path) -> per-event poison: attempts incremented; status set to failed or dead (when attempts >= max_attempts).

• Any other exception (publish_batch path) -> whole-batch poison: all rows in the claimed batch are poisoned together (not row-by-row).

E.g.:

import asyncio
from alchemiq.outbox import Relay

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

# run in a background task:
task = asyncio.create_task(relay.run())

# ... on shutdown:
relay.stop()
await task

Parameters:

• publisher (Publisher) – delivery adapter satisfying the Publisher protocol.

• batch_size (int) – maximum rows claimed per cycle (default 100).

• poll_interval (float) – seconds to wait between cycles when the batch was not full (default 1.0).

• max_attempts (int) – delivery attempts before a row is marked dead (default 5).

• error_backoff (float) – seconds to sleep after a transient broker error (default 5.0).

See also

Publisher - the delivery protocol. publish() - manual event emission outside of a model signal.

stop()

Signal the relay loop to stop after the current cycle completes.

Return type:

None

async run()

Start the relay polling loop; blocks until stop() is called or cancelled.

Spawns no threads. Designed to run as an asyncio task. Cancellation is handled gracefully: the stopping event is set and CancelledError is re-raised.

Note

Requires alchemiq to be configured (configure called) before run() is awaited.

Return type:

None

async alchemiq.publish(topic, payload, *, key=None, headers=None)

Write an outbox event row in its own autocommit transaction.

Use this for manual event emission (e.g. from a background task or CLI) when no model signal is firing. To tie the event to a business transaction, open a UnitOfWork around the call.

E.g.:

from alchemiq import publish

await publish("user.signed_up", {"id": 1, "email": "ada@x.io"})

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

Parameters:

• topic (str) – broker routing key (e.g. "user.signed_up").

• payload (dict[str, Any] | BaseModel) – event data - a plain dict or a Pydantic BaseModel.

• key (str | None) – stored as aggregate_id; typically used as the broker partition key.

• headers (dict[str, Any] | None) – optional broker-level headers dict.

Return type:

None

See also

Relay - background worker that picks up and delivers these rows.

exception alchemiq.PublishError

Bases: Exception

Base exception for publisher delivery errors.

exception alchemiq.TransientPublishError

Bases: PublishError

Broker unreachable or connection lost.

When raised by a Publisher, the Relay rolls back the whole batch without incrementing attempts and sleeps error_backoff seconds before retrying. Use this for temporary infrastructure failures, not logical errors.