python-cqrs

As a service grows, business logic spreads across routes, consumers, and workers. python-cqrs puts each use case in one handler behind a mediator. Routes and consumers only send requests, and side effects become events with their own handlers.

Once services talk to each other, a database write and a broker message are no longer atomic. The outbox and inbox keep messages from being lost or handled twice, and a saga undoes completed steps when a later one fails.

Dataclasses or Pydantic · SQLAlchemy or SQLModel · Kafka or RabbitMQ · FastAPI and FastStream · Python 3.10+

Installation

pip install python-cqrs
pip install "python-cqrs[pydantic,sqlalchemy,kafka]"  # add what you use

Extras: pydantic, sqlalchemy, sqlmodel, kafka, rabbit, protobuf, aiobreaker, dishka, testing. Details: installation. From 4.x: migration guide.

Quick start

import di
import cqrs
from cqrs.bootstrap import requests as bootstrap


class OrderCreated(cqrs.Response):
    order_id: str


class CreateOrder(cqrs.Request[OrderCreated]):
    sku: str
    amount: float


class CreateOrderHandler(cqrs.RequestHandler[CreateOrder]):
    async def handle(self, request: CreateOrder):
        return OrderCreated(order_id=f"ord-{request.sku}")


def commands_mapper(mapper: cqrs.RequestMap) -> None:
    mapper.bind(CreateOrder, CreateOrderHandler)


mediator = bootstrap.bootstrap(di_container=di.Container(), commands_mapper=commands_mapper)
result = await mediator.send(CreateOrder(sku="book", amount=99.99))  # OrderCreated(order_id="ord-book")

result is typed as OrderCreated. Queries use the same shape. Handler dependencies come from di, dependency-injector, or dishka. Request handlers · Bootstrap · Dependency injection

Saga

Paying for an order touches the warehouse service and the payment service, and no transaction spans both. If the charge fails, the stock stays reserved. A saga runs the steps in order and, on failure, calls compensate on the completed ones in reverse.

import dataclasses
from cqrs.bootstrap import saga as saga_bootstrap
from cqrs.saga.models import SagaContext


@dataclasses.dataclass
class OrderContext(SagaContext):
    order_id: str
    amount: float
    reservation_id: str | None = None


class ReserveStock(cqrs.SagaStepHandler[OrderContext, None]):
    def __init__(self, inventory: InventoryClient) -> None:
        self._inventory = inventory

    async def act(self, context: OrderContext):
        context.reservation_id = await self._inventory.reserve(context.order_id)
        return self._generate_step_result(None)

    async def compensate(self, context: OrderContext) -> None:
        await self._inventory.release(context.reservation_id)


class ChargeCard(cqrs.SagaStepHandler[OrderContext, None]):
    def __init__(self, payments: PaymentClient) -> None:
        self._payments = payments

    async def act(self, context: OrderContext):
        await self._payments.charge(context.order_id, context.amount)
        return self._generate_step_result(None)

    async def compensate(self, context: OrderContext) -> None:
        await self._payments.refund(context.order_id)


class OrderSaga(cqrs.Saga[OrderContext]):
    steps = [ReserveStock, ChargeCard]


def sagas_mapper(mapper: cqrs.SagaMap) -> None:
    mapper.bind(OrderContext, OrderSaga)


mediator = saga_bootstrap.bootstrap(di_container=di.Container(), sagas_mapper=sagas_mapper)

# if ChargeCard raises, ReserveStock.compensate releases the stock
async for step in mediator.execute(OrderContext(order_id="ord-1", amount=99.99)):
    print(step.step_type.__name__)

Storage, recovery after a crash, and fallback steps: Saga.

Outbox and inbox

Outbox

Commit the order, then publish to Kafka: a crash between the two loses the event. The outbox writes the event in the same transaction as the order. A publisher sends it later, at least once.

class OrderPlaced(cqrs.NotificationEvent):
    event_name: str = "order_placed"
    topic: str = "orders"


class CreateOrderHandler(cqrs.RequestHandler[CreateOrder]):
    def __init__(self, outbox: cqrs.OutboxedEventRepository) -> None:
        self._outbox = outbox

    async def handle(self, request: CreateOrder):
        order_id = f"ord-{request.sku}"
        # write the order row through the same session here
        self._outbox.add(OrderPlaced(payload={"order_id": order_id}))
        await self._outbox.commit()
        return OrderCreated(order_id=order_id)

Outbox · Database support · examples/outbox/

Inbox

The broker redelivers, and a consumer can crash halfway through a handler. The FastStream subscriber only stores the message. A repeated payment_id is skipped.

class ConfirmPayment(cqrs.Request[None]):
    order_id: str


@broker.subscriber("payments")
async def on_payment(
    body: dict,
    inbox: cqrs.InboxRepository = Depends(inbox_repository),  # commits after return, then ack
) -> None:
    await inbox.add([(body["payment_id"], ConfirmPayment(order_id=body["order_id"]))])

A worker runs stored rows through the mediator. One commit saves the handler's changes and the processed status.

processor = cqrs.InboxProcessor(inbox, message_map, request_mediator=mediator)
for row in await inbox.get_many():
    await processor.process(row)
await session.commit()

Inbox · examples/inbox/

Domain events

A placed order also updates a read model and sends an email. In the command handler, that couples unrelated work. Instead, the handler emits an event, and each reaction is its own handler.

class OrderPlaced(cqrs.DomainEvent):
    order_id: str


class CreateOrderHandler(cqrs.RequestHandler[CreateOrder]):
    def __init__(self) -> None:
        self._events: list[cqrs.Event] = []

    @property
    def events(self) -> list[cqrs.Event]:
        return self._events

    async def handle(self, request: CreateOrder):
        order_id = f"ord-{request.sku}"
        self._events.append(OrderPlaced(order_id=order_id))
        return OrderCreated(order_id=order_id)


class UpdateOrderList(cqrs.EventHandler[OrderPlaced]):
    async def handle(self, event: OrderPlaced) -> None: ...


class SendConfirmation(cqrs.EventHandler[OrderPlaced]):
    async def handle(self, event: OrderPlaced) -> None: ...


def domain_events_mapper(mapper: cqrs.EventMap) -> None:
    mapper.bind(OrderPlaced, UpdateOrderList)
    mapper.bind(OrderPlaced, SendConfirmation)

Pass domain_events_mapper to bootstrap. Event handlers run after the command, in parallel when enabled. Events · Parallel processing · Scoped dependencies

FastAPI

The route parses HTTP and sends the request. The handler does not import FastAPI.

@app.post("/orders")
async def create_order(
    command: CreateOrder,
    mediator: cqrs.RequestMediator = Depends(mediator_factory),
) -> dict[str, str]:
    result = await mediator.send(command)
    return {"order_id": result.order_id}

FastStream consumers and SSE streaming work the same way. FastAPI · FastStream · SSE streaming

Testing

Mocks show that a method was called, not what the service did. test_harness runs the real mediators and records what was handled, published, and stored.

from cqrs.testing import test_harness


async def test_create_order() -> None:
    # Arrange
    async with test_harness(di.Container(), commands_mapper=commands_mapper) as harness:

        # Act
        result = await harness.mediator.send(CreateOrder(sku="book", amount=99.99))

        # Assert
        assert result.order_id == "ord-book"
        assert harness.requests.contains(CreateOrder)
        harness.invocations.assert_entered(CreateOrderHandler)
        harness.published.assert_empty()

The same journals cover domain events, the outbox, and saga compensation. Testing · examples/testing/

When NOT to use python-cqrs

CQRS, Outbox, and Saga are tools for specific problems — not a default for every service.

• Prefer a local DB transaction for single-service CRUD (one database, no broker, no compensation).
• Use Transactional Outbox when you dual-write to a database and a message broker. Commit the business row and the outbox row together; a publisher drains the outbox later.
• Use Saga when a multi-step flow crosses service/database boundaries and a later step must be compensated if it fails.

Do not add a Saga for a single UPDATE in one service, and do not use a Saga as a substitute for Outbox. Full write-up: When NOT to use python-cqrs.

Documentation

Full guides: mkdocs.python-cqrs.dev. Also there:

• Request and response types: dataclasses, Pydantic, msgspec, attrs, and others
• Streaming: handlers that yield results as they finish
• Chain of responsibility: several handlers for one request, the first match wins
• Saga storage and recovery: resume a saga after a crash
• Fallback and circuit breaker: a spare step or handler when the primary fails
• Middlewares and message brokers
• Protobuf and Mermaid diagrams

Agent skills

Skills tell a coding agent which feature fits the task. Install from your service's root:

npx skills add pypatterns/python-cqrs --skill cqrs-setup -y
npx skills add pypatterns/python-cqrs --skill cqrs-eda -y
npx skills add pypatterns/python-cqrs --skill cqrs-saga -y
npx skills add pypatterns/python-cqrs --skill cqrs-streaming -y
npx skills add pypatterns/python-cqrs --skill cqrs-cor -y
npx skills add pypatterns/python-cqrs --skill cqrs-testing -y

Also copy templates/AGENTS.md into the application. Agent skills · Rules for agents

Contributing

pip install -e ".[dev]"
pytest -c ./tests/pytest-config.ini ./tests/unit
pre-commit install

Details: CONTRIBUTING.md. Vulnerabilities: SECURITY.md.

License

MIT