Outbox

• Start here: Outbox guide
• Common recipes: publish, @handler, relay
• Configuration: Backend, Configuration

grelmicro.outbox

Outbox.

Run an async handler exactly after your database transaction commits, at least once. Stage a message inside your own transaction with publish, then a background relay delivers it to a registered @handler.

Outbox

Outbox(
    source: Provider
    | OutboxBackend
    | type[Provider | OutboxBackend],
    *,
    name: str = "default",
    config: OutboxConfig | None = None,
    relay: bool | None = None,
    table: str | None = None,
    poll_interval: float | None = None,
    batch_size: int | None = None,
    lease_duration: float | None = None,
    max_attempts: int | None = None,
    retry_base: float | None = None,
    retry_max: float | None = None,
    retry_jitter: float | None = None,
    concurrency: int | None = None,
    dead_letter: bool | None = None,
    keep_delivered: bool | timedelta | None = None,
    auto_migrate: bool | None = None,
    notify: bool | None = None,
    env_load: bool | None = None,
    shutdown_timeout: float = 30.0,
)

Outbox component: stages messages and runs their handlers.

Registered as micro.outbox after Grelmicro(uses=[Outbox(...)]). Accepts a Provider or an OutboxBackend. When given a Provider, the component calls provider.outbox() to build the matching adapter.

Example

from grelmicro import Grelmicro
from grelmicro.outbox import Message, Outbox
from grelmicro.providers.postgres import PostgresProvider

postgres = PostgresProvider("postgresql://localhost:5432/app")
outbox = Outbox(postgres)


@outbox.handler("email.welcome")
async def send_welcome(message: Message) -> None:
    await mailer.send(to=message.payload["to"])


micro = Grelmicro(uses=[outbox])

Read more in the Outbox docs.

Initialize the component and resolve its config and backend.

PARAMETER	DESCRIPTION
source	A Provider (e.g. PostgresProvider) or an OutboxBackend. When a Provider is given, the component calls provider.outbox() to build the matching adapter. TYPE: Provider | OutboxBackend | type[Provider | OutboxBackend]
name	Registration name. TYPE: str DEFAULT: 'default'
config	A pre-built OutboxConfig. Mutually exclusive with kwargs. TYPE: OutboxConfig | None DEFAULT: None
relay	Run the relay on this replica. TYPE: bool | None DEFAULT: None
table	Table that stores messages. TYPE: str | None DEFAULT: None
poll_interval	Seconds between fallback polls. TYPE: float | None DEFAULT: None
batch_size	Claim ceiling per cycle. TYPE: int | None DEFAULT: None
lease_duration	Seconds a claimed message stays invisible. TYPE: float | None DEFAULT: None
max_attempts	Attempts before dead-lettering. TYPE: int | None DEFAULT: None
retry_base	Base backoff in seconds. TYPE: float | None DEFAULT: None
retry_max	Maximum backoff in seconds. TYPE: float | None DEFAULT: None
retry_jitter	Jitter fraction applied to the backoff. TYPE: float | None DEFAULT: None
concurrency	Maximum handlers running at once. TYPE: int | None DEFAULT: None
dead_letter	Move exhausted messages to the dead state. TYPE: bool | None DEFAULT: None
keep_delivered	Keep delivered rows instead of deleting them. A timedelta keeps them for that long, then the relay purges them. TYPE: bool | timedelta | None DEFAULT: None
auto_migrate	Create the table on first connect. TYPE: bool | None DEFAULT: None
notify	Use LISTEN/NOTIFY for low-latency wakeups. TYPE: bool | None DEFAULT: None
env_load	Read missing settings from the environment. TYPE: bool | None DEFAULT: None
shutdown_timeout	Seconds to let in-flight handlers drain on shutdown. TYPE: float DEFAULT: 30.0

kind class-attribute

kind: str = 'outbox'

name property

name: str

Return the registration name.

config property

config: OutboxConfig

Return the resolved configuration.

backend property

backend: OutboxBackend

The underlying OutboxBackend.

current classmethod

current(name: str = 'default') -> Outbox

Return the registered Outbox from the active Grelmicro app.

Lets a producer publish without holding the constructed instance or a config-bound module singleton:

await Outbox.current().publish(conn, WelcomeEmail(to=email))

PARAMETER	DESCRIPTION
name	Registration name. TYPE: str DEFAULT: 'default'

RAISES	DESCRIPTION
OutOfContextError	No active app, or no Outbox registered under name. Run inside async with micro: or after micro.install(app), with an Outbox in uses=[...].

handler

handler(
    target: type[Any] | str, *, topic: str | None = None
) -> Callable[
    [Callable[[Message[Any]], Awaitable[None]]],
    Callable[[Message[Any]], Awaitable[None]],
]

Register an async handler for a payload model or a topic.

PARAMETER	DESCRIPTION
target	A payload model or a topic string to bind the handler to. TYPE: type[Any] | str
topic	Override the derived topic. TYPE: str | None DEFAULT: None

publish async

publish(
    handle: object,
    target: object,
    payload: Mapping[str, Any] | None = None,
    *,
    key: str | None = None,
    headers: Mapping[str, Any] | None = None,
    dedup_key: str | None = None,
    delay: float | timedelta | None = None,
) -> bool

Stage a message inside the caller's transaction.

Returns True when the message is staged, False when a dedup_key collides and the message is skipped.

PARAMETER	DESCRIPTION
handle	Your connection or session, already inside a transaction. TYPE: object
target	A payload model instance, or a topic string with payload. TYPE: object
payload	The payload dict when target is a topic string. TYPE: Mapping[str, Any] | None DEFAULT: None
key	Ordering or partition key. TYPE: str | None DEFAULT: None
headers	Metadata to carry with the message. TYPE: Mapping[str, Any] | None DEFAULT: None
dedup_key	Producer-side deduplication key. TYPE: str | None DEFAULT: None
delay	Hold the message back for this long before delivery. TYPE: float | timedelta | None DEFAULT: None

redrive async

redrive(*, topic: str | None = None) -> int

Move dead messages back to pending. Returns the count moved.

purge async

purge(
    *, older_than: timedelta | float | None = None
) -> int

Delete delivered and dead rows. Returns the count removed.

Use it to trim the table once delivered or dead messages are no longer needed. Pending and in-flight messages are never touched.

PARAMETER	DESCRIPTION
older_than	Only purge terminal rows older than this. None purges all. TYPE: timedelta | float | None DEFAULT: None

OutboxBackend

Bases: Protocol

Protocol for outbox storage backends.

enqueue async

enqueue(handle: object, record: OutboxRecord) -> bool

Stage a record inside the caller's transaction.

Returns False when a dedup_key collides and the row is skipped, True when the row is inserted.

PARAMETER	DESCRIPTION
handle	The caller's open connection or session. The insert joins it. TYPE: object
record	The message to stage. TYPE: OutboxRecord

claim async

claim(
    *, topics: Sequence[str], limit: int, lease: float
) -> list[OutboxRecord]

Claim up to limit due messages for the given topics.

PARAMETER	DESCRIPTION
topics	Topics with a registered handler. Only these are claimed. TYPE: Sequence[str]
limit	Maximum number of messages to claim. TYPE: int
lease	Seconds the claimed messages stay invisible. TYPE: float

complete async

complete(
    *, message_id: UUID, attempts: int, keep: bool
) -> None

Mark a message delivered.

PARAMETER	DESCRIPTION
message_id	The delivered message id. TYPE: UUID
attempts	The claimed attempt count, fencing a stale relay's write. TYPE: int
keep	Keep the row in the delivered state instead of deleting it. TYPE: bool

reschedule async

reschedule(
    *,
    message_id: UUID,
    attempts: int,
    delay: float,
    error: str,
    dead: bool,
) -> None

Reschedule a failed message or dead-letter it.

PARAMETER	DESCRIPTION
message_id	The message id to reschedule. TYPE: UUID
attempts	The claimed attempt count, fencing a stale relay's write. TYPE: int
delay	Seconds until the next attempt. TYPE: float
error	The last handler error. TYPE: str
dead	Move the message to the dead state instead of retrying. TYPE: bool

purge async

purge(
    *,
    before_seconds: float | None = None,
    states: tuple[Literal["delivered", "dead"], ...] = (
        "delivered",
        "dead",
    ),
) -> int

Delete terminal rows in the given states. Returns the count removed.

PARAMETER	DESCRIPTION
before_seconds	Only purge terminal rows older than this many seconds. TYPE: float | None DEFAULT: None
states	Terminal states to purge. TYPE: tuple[Literal['delivered', 'dead'], ...] DEFAULT: ('delivered', 'dead')

redrive async

redrive(*, topic: str | None = None) -> int

Move dead messages back to pending. Returns the count moved.

PARAMETER	DESCRIPTION
topic	Restrict to one topic. None redrives every dead message. TYPE: str | None DEFAULT: None

wait_notify async

wait_notify(*, timeout: float) -> None

Return when a new message is signalled or timeout elapses.

PARAMETER	DESCRIPTION
timeout	Maximum seconds to wait for a wake. TYPE: float

OutboxConfig

Bases: BaseModel

Outbox settings.

Plain BaseModel (env-free). Component defaults resolve from the environment under GREL_OUTBOX_ unless fields are set directly.

table class-attribute instance-attribute

table: str = 'grelmicro_outbox'

Table that stores staged messages.

relay class-attribute instance-attribute

relay: bool = True

Run the background relay on this replica.

poll_interval class-attribute instance-attribute

poll_interval: float = 1.5

Seconds between fallback polls. The NOTIFY wake handles the fast path.

batch_size class-attribute instance-attribute

batch_size: int = 100

Claim ceiling per cycle, capped by free handler slots.

lease_duration class-attribute instance-attribute

lease_duration: float = 30

Seconds a claimed message stays invisible. Handlers must finish within it.

max_attempts class-attribute instance-attribute

max_attempts: int = 10

Attempts before a message is dead-lettered.

retry_base class-attribute instance-attribute

retry_base: float = 1

Base backoff in seconds.

retry_max class-attribute instance-attribute

retry_max: float = 300

Maximum backoff in seconds.

retry_jitter class-attribute instance-attribute

retry_jitter: float = 1

Jitter fraction applied to the backoff, from 0 to 1.

concurrency class-attribute instance-attribute

concurrency: int = 50

Maximum handlers running at once in each relay.

dead_letter class-attribute instance-attribute

dead_letter: bool = True

Move a message to the dead state after max_attempts. When False, a failing message is retried forever on the backoff.

keep_delivered class-attribute instance-attribute

keep_delivered: bool | timedelta = False

Keep delivered rows instead of deleting them. Pass a timedelta to keep them for that long, then the relay purges them.

auto_migrate class-attribute instance-attribute

auto_migrate: bool = True

Create the table on first connect.

notify class-attribute instance-attribute

notify: bool = True

Use LISTEN/NOTIFY for low-latency wakeups. Disable behind PgBouncer.

Message dataclass

Message(
    id: UUID,
    topic: str,
    key: str | None,
    data: T | None,
    payload: Mapping[str, Any],
    headers: Mapping[str, Any],
    attempts: int,
)

A staged message delivered to a handler.

data holds the validated payload model for a typed handler and is None for a topic handler. payload holds the raw payload dict for both. Use id as the idempotency key for the side effect.

id instance-attribute

id: UUID

topic instance-attribute

topic: str

key instance-attribute

key: str | None

data instance-attribute

data: T | None

payload instance-attribute

payload: Mapping[str, Any]

headers instance-attribute

headers: Mapping[str, Any]

attempts instance-attribute

attempts: int

Retry

Retry(*, delay: float | timedelta | None = None)

Bases: Exception

Raise from a handler to reschedule the message.

The relay reschedules the message after delay, or on the configured backoff when delay is None. The attempt still counts toward max_attempts.

Initialize the signal.

PARAMETER	DESCRIPTION
delay	Time until the next attempt. None uses the backoff. TYPE: float | timedelta | None DEFAULT: None

delay instance-attribute

delay = delay

Cancel

Cancel(reason: str)

Bases: Exception

Raise from a handler to dead-letter the message now.

The relay moves the message to the dead state without further attempts and records reason as its last error.

Initialize the signal.

PARAMETER	DESCRIPTION
reason	Why the message is dead-lettered. Stored as the last error. TYPE: str

reason instance-attribute

reason = reason

OutboxError

Bases: GrelmicroError

Base outbox error.

OutboxHandleError

Bases: OutboxError, TypeError

Raised when publish gets a handle it cannot use.

A pool or an engine hands out a fresh connection, so the message would land in a separate transaction. Pass a connection or session that is already inside your transaction.

OutboxTransactionError

Bases: OutboxError, RuntimeError

Raised when publish gets a handle with no open transaction.

The message must join the caller's transaction so it commits with the business write. A connection or session outside a transaction would commit the message on its own, so publish refuses it.

HandlerNotFoundError

Bases: OutboxError, LookupError

Raised when a topic has no registered handler.

HandlerAlreadyRegisteredError

Bases: OutboxError, ValueError

Raised when a topic is registered twice.