FastStream integration¶

FastStream integration for alchemiq. Requires the [faststream] extra.

class alchemiq.faststream.FastStreamPublisher(broker, *, transient_errors=(<class 'ConnectionError'>, <class 'OSError'>, <class 'TimeoutError'>))[source]¶

Bases: object

Publish OutboxMessage objects to any FastStream broker.

Implements the Publisher protocol so it can be wired directly into a Relay. Works with any FastStream broker (RabbitMQ, Kafka, NATS, Redis) or any object that exposes the same duck-typed publish signature:

async def publish(message, destination, *, headers, correlation_id) -> None: ...

The caller owns the broker’s connection lifecycle (connect / start / stop / close). correlation_id is set to str(message.id) - the dedup key consumers use under the at-least-once delivery contract.

Connection-class errors (the transient_errors tuple) are re-raised as TransientPublishError so the relay backs off without burning retry attempts. Pass broker-specific connection exception classes in transient_errors without importing them in this module.

E.g.:

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

_broker = RabbitBroker()
relay = Relay(FastStreamPublisher(_broker), batch_size=10)

# custom transient errors:
publisher = FastStreamPublisher(_broker, transient_errors=(BrokerGone,))

Parameters:

broker (Any) – a started FastStream broker or duck-typed equivalent.
transient_errors (tuple[type[Exception], ...]) – exception classes that indicate a transient connection failure and should be wrapped in TransientPublishError; defaults to (ConnectionError, OSError, TimeoutError).

See also

Relay - drives the outbox drain loop using this publisher.

async publish(message)[source]¶

Publish message to its topic via the broker.

Sets correlation_id to str(message.id) and forwards all alchemiq.* metadata headers present on the message. Re-raises connection-class exceptions as TransientPublishError so the relay backs off without burning retry attempts.

Parameters:: message (OutboxMessage) – the OutboxMessage to deliver.
Raises:: TransientPublishError – on connection-class errors listed in transient_errors.
Return type:: None

alchemiq.faststream.lifespan(dsn, *, create_all=False, **engine_kwargs)[source]¶

Build a FastStream lifespan that owns the alchemiq engine lifecycle.

Calls configure() on startup and dispose() on shutdown. The returned factory absorbs whatever context argument FastStream passes (or none), so it is compatible with all FastStream broker and app variants.

E.g.:

from faststream import FastStream
from faststream.rabbit import RabbitBroker
from alchemiq.faststream import lifespan

broker = RabbitBroker("amqp://guest:guest@localhost/")
app = FastStream(broker, lifespan=lifespan("postgresql+asyncpg://u:p@h/db"))

# create tables on startup (dev/test only):
app = FastStream(broker, lifespan=lifespan(dsn, create_all=True))

Parameters:

dsn (str) – SQLAlchemy async database URL (e.g. "postgresql+asyncpg://user:pw@host/db").
create_all (bool) – when True, run CREATE TABLE IF NOT EXISTS for all registered models before the broker starts consuming.
engine_kwargs (Any) – extra keyword arguments forwarded to create_async_engine (e.g. pool_size=5).

Returns:

an async context-manager factory compatible with FastStream(lifespan=...).

Return type:

Callable[[…], AbstractAsyncContextManager[None]]