Implementation¶
-
Back to Transactional Outbox Overview
Return to the Transactional Outbox overview page with all topics.
Overview¶
The OutboxedEventRepository interface defines the contract for outbox implementations:
from abc import ABC
from typing import Generic, List, Text
import cqrs
from cqrs.outbox.repository import EventStatus, OutboxedEvent
class OutboxedEventRepository(ABC, Generic[Session]):
"""Abstract interface for outbox event repositories."""
def add(self, event: cqrs.NotificationEvent) -> None:
"""Add an event to the outbox repository."""
pass
async def get_many(
self,
batch_size: int = 100,
topic: Text | None = None,
) -> List[OutboxedEvent]:
"""Get many events from the repository."""
pass
async def update_status(
self,
outboxed_event_id: int,
new_status: EventStatus,
) -> None:
"""Update the event status."""
pass
async def commit(self) -> None:
"""Commit the transaction."""
pass
async def rollback(self) -> None:
"""Rollback the transaction."""
pass
Methods¶
add(event)¶
Adds an event to the outbox. The event is stored but not yet committed. Must be called within a transaction context.
Parameters:
- event — cqrs.NotificationEvent to store
Example:
outbox.add(
cqrs.NotificationEvent[UserJoinedPayload](
event_name="user_joined",
topic="user_events",
payload=UserJoinedPayload(user_id="123", meeting_id="456"),
)
)
get_many(batch_size, topic)¶
Retrieves events from the outbox in batches. Used by the publisher process.
Parameters:
- batch_size — Maximum number of events to retrieve (default: 100)
- topic — Optional topic filter to retrieve events for specific topic
Returns:
- List[OutboxedEvent] — List of outboxed events
Example:
update_status(outboxed_event_id, new_status)¶
Updates the status of an event in the outbox. Used to mark events as produced or failed.
Parameters:
- outboxed_event_id — ID of the event in outbox
- new_status — New status (EventStatus.NEW, EventStatus.PRODUCED, EventStatus.NOT_PRODUCED)
Example:
commit()¶
Commits the current transaction. All events added via add() are persisted.
Example:
rollback()¶
Rolls back the current transaction. All events added via add() are discarded.
Example:
try:
outbox.add(event1)
await outbox.commit()
except Exception:
await outbox.rollback() # Event is discarded
Events in the outbox have three possible statuses:
class EventStatus(StrEnum):
NEW = "new" # Event is new and ready to be published
PRODUCED = "produced" # Event has been successfully published
NOT_PRODUCED = "not_produced" # Event publishing failed
Stored values are the member names
The column is mapped with sqlalchemy.Enum(EventStatus), which persists the names of the enum members.
In the database you will see NEW, PRODUCED and NOT_PRODUCED — not the lowercase Python values. This
matters when you query the table with raw SQL or write migrations by hand.
The python-cqrs package includes a SQLAlchemy implementation of the outbox pattern. From 5.1 the same repository behavior is also available with SQLModel. The table, dialect-aware column types, batch query, flush budget, and status ordering are the same; pick one ORM per process. See ORM.
Database Schema¶
The outbox model is built from dialect-aware column types, so the rendered DDL differs between databases. The two supported variants are:
MySQL / MariaDB:
CREATE TABLE outbox (
id BIGINT NOT NULL AUTO_INCREMENT,
event_id BINARY(16) NOT NULL,
event_id_bin BINARY(16) NOT NULL,
event_status ENUM('NEW', 'PRODUCED', 'NOT_PRODUCED') NOT NULL,
flush_counter SMALLINT NOT NULL,
event_name VARCHAR(255) NOT NULL,
topic VARCHAR(255) NOT NULL,
created_at DATETIME NOT NULL DEFAULT now(),
payload BLOB NOT NULL,
PRIMARY KEY (id),
CONSTRAINT event_id_unique_index UNIQUE (event_id_bin, event_name)
);
PostgreSQL:
CREATE TYPE eventstatus AS ENUM ('NEW', 'PRODUCED', 'NOT_PRODUCED');
CREATE TABLE outbox (
id BIGINT GENERATED BY DEFAULT AS IDENTITY,
event_id UUID NOT NULL,
event_id_bin BYTEA NOT NULL,
event_status eventstatus NOT NULL,
flush_counter SMALLINT NOT NULL,
event_name VARCHAR(255) NOT NULL,
topic VARCHAR(255) NOT NULL,
created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT now() NOT NULL,
payload BYTEA NOT NULL,
PRIMARY KEY (id),
CONSTRAINT event_id_unique_index UNIQUE (event_id_bin, event_name)
);
Other databases and Alembic
See Database Support for the full support matrix, the type mapping per dialect, the Alembic recipe and how to register a native type for a database the library does not know about.
Features¶
- UUID Support — Events have unique IDs for idempotency
- Binary Storage — Efficient storage of UUIDs as binary
- Portable Schema — Native column types per dialect (MySQL, MariaDB, PostgreSQL) with a portable fallback
- Status Tracking — Tracks event publishing status
- Compression Support — Optional payload compression
- Batch Processing — Optimized queries for batch retrieval
Usage¶
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from cqrs import SqlAlchemyOutboxedEventRepository
from cqrs.compressors import ZlibCompressor
# Create session factory (MySQL)
session_factory = async_sessionmaker(
create_async_engine(
"mysql+asyncmy://user:password@localhost/database",
isolation_level="REPEATABLE READ",
)
)
# ...or PostgreSQL — only the DSN changes
session_factory = async_sessionmaker(
create_async_engine(
"postgresql+asyncpg://user:password@localhost/database",
isolation_level="REPEATABLE READ",
)
)
# Create repository with optional compression
outbox = SqlAlchemyOutboxedEventRepository(
session=session_factory(),
compressor=ZlibCompressor(), # Optional
)
# Use in command handler
class MyCommandHandler(RequestHandler[MyCommand, None]):
def __init__(self, outbox: OutboxedEventRepository):
self.outbox = outbox
async def handle(self, request: MyCommand) -> None:
# Business logic
...
# Save events to outbox
self.outbox.add(
cqrs.NotificationEvent[MyPayload](
event_name="my_event",
topic="my_topic",
payload=MyPayload(...),
)
)
# Commit transaction
await self.outbox.commit()
Configuration¶
Table Name¶
By default, the table name is outbox. To rename it — or to attach the outbox model to your own declarative
base so that it shares one MetaData with the rest of your schema — use rebind_outbox_model:
import cqrs
from cqrs.outbox.sqlalchemy import OutboxModel
from myapp.orm import Base # your own declarative base
cqrs.rebind_outbox_model(OutboxModel, Base, table_name="my_outbox_table")
Call it before the table is used
rebind_outbox_model mutates the model and re-registers its table in the target metadata, so call it at
import/startup time — before create_all(), before generating migrations and before the first query.
Compression¶
Compression is optional but recommended for large payloads: