Python CQRS: Simplify distributed systems with Sagas and Outbox

Distributed systems often introduce complexity that overshadows the business logic they’re meant to support. Managing consistency across microservices becomes a challenge when a single transaction in one service doesn’t automatically align with others. This is where Command Query Responsibility Segregation (CQRS) and event-driven architectures step in, offering a structured approach to handling distributed workflows without reinventing fundamental patterns.

Python developers now have a streamlined tool to implement these patterns efficiently. The open-source library python-cqrs provides a lightweight framework for building robust distributed systems in Python, combining CQRS, sagas, and transactional outbox patterns into a cohesive architecture. Whether you’re managing financial transactions, inventory updates, or multi-service workflows, this approach ensures reliability without excessive boilerplate.

How python-cqrs streamlines distributed workflows

python-cqrs evolved from a fork of the diator library but has since matured into its own framework, specifically designed to simplify the implementation of CQRS and event-driven systems in Python. The core philosophy is to separate command and query operations while ensuring that every action within a distributed system remains consistent and recoverable, even in the face of failures.

At its heart, python-cqrs leverages a mediator pattern to decouple command handling from the underlying infrastructure. This means business logic remains isolated from transport layers like HTTP, Kafka, or CLI, making the codebase more maintainable and easier to test.

Here’s a minimal example of how commands and handlers are structured in python-cqrs:

import di
import cqrs
from cqrs.requests import bootstrap

# Define a command with input parameters
class CreateUserCommand(cqrs.Request):
    email: str
    name: str

# Define the expected response
class CreateUserResponse(cqrs.Response):
    user_id: str

# Implement a pure handler with no infrastructure dependencies
class CreateUserHandler(cqrs.RequestHandler[CreateUserCommand, CreateUserResponse]):
    async def handle(self, request: CreateUserCommand) -> CreateUserResponse:
        user_id = f"user_{request.email}"
        return CreateUserResponse(user_id=user_id)

# Set up the mediator with dependency injection
mediator = bootstrap.bootstrap(
    di_container=di.Container(),
    commands_mapper=lambda m: m.bind(CreateUserCommand, CreateUserHandler),
)

# Execute the command
result = await mediator.send(
    CreateUserCommand(email="user@example.com", name="John"),
)

This structure ensures that commands and queries are processed consistently, regardless of the underlying transport or infrastructure.

Managing distributed transactions with Sagas

In distributed systems, a single operation often involves multiple services. For example, processing an order might require reserving inventory, charging payment, and sending a confirmation email. If one step fails, the entire operation must be rolled back to maintain consistency.

Sagas provide a solution by breaking complex workflows into smaller, manageable steps. Each step is executed as a separate command, and if any step fails, compensating actions are triggered to undo previous steps. python-cqrs supports sagas with persisted state, automatic compensation, and recovery mechanisms, ensuring workflows remain reliable even after system crashes.

The framework offers fallback and circuit-style options for sagas, allowing developers to define recovery strategies based on the nature of the failure. This is particularly useful in financial systems or inventory management, where consistency is critical.

Ensuring reliable event delivery with the Transactional Outbox

A common pitfall in event-driven architectures is the "committed in database but never reached the broker" scenario. When a service writes to its database but the message to Kafka (or another broker) fails, the system becomes inconsistent. The transactional outbox pattern solves this by storing events in the same database transaction as the state changes.

With python-cqrs, integration events are written to an outbox table within the same transaction as the primary data writes. A separate process then reads from this table and publishes events to the broker, ensuring at-least-once delivery. Consumers should be designed to handle duplicate messages, typically through idempotency keys or deduplication logic.

Here’s a simplified example of how the outbox pattern is implemented:

import cqrs
from pydantic import BaseModel

# Define the payload for an integration event
class UserJoinedPayload(BaseModel, frozen=True):
    user_id: str
    meeting_id: str

# Register the event type and its wire name
cqrs.OutboxedEventMap.register(
    "user_joined",  # Topic or routing key
    cqrs.NotificationEvent[UserJoinedPayload],
)

In a command handler, you append a NotificationEvent to the outbox repository within the same transaction as the aggregate update. A background process then publishes these events to the broker, ensuring reliable delivery.

Separating in-memory and integration events

Not all events need to be published to external systems. Some events are purely internal and used to update read models or trigger in-memory logic. python-cqrs distinguishes between domain events (in-process) and notification events (integration).

Domain events are dispatched immediately within the same process and are ideal for updating read models or triggering side effects that don’t require durability across service boundaries. For example:

import cqrs
import di
from cqrs.requests import bootstrap

class CreateOrder(cqrs.Request):
    order_id: str

class OrderLineAdded(cqrs.DomainEvent):
    order_id: str
    sku: str

class CreateOrderHandler(cqrs.RequestHandler[CreateOrder, cqrs.Response]):
    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) -> cqrs.Response:
        self._events.append(
            OrderLineAdded(order_id=request.order_id, sku="sku-1"),
        )
        return cqrs.Response()

class OnOrderLineAdded(cqrs.EventHandler[OrderLineAdded]):
    async def handle(self, event: OrderLineAdded) -> None:
        # Update a local projection or cache
        pass

# Configure the mediator with event handlers
mediator = bootstrap.bootstrap(
    di_container=di.Container(),
    commands_mapper=lambda m: m.bind(CreateOrder, CreateOrderHandler),
    domain_events_mapper=lambda m: m.bind(OrderLineAdded, OnOrderLineAdded),
    max_concurrent_event_handlers=4,  # Control concurrency
)

This separation ensures that internal workflows remain fast and in-memory, while integration events are handled reliably and durably.

Building for the future of distributed systems

As microservices and distributed architectures continue to evolve, the need for reliable, maintainable patterns becomes even more critical. Libraries like python-cqrs provide a practical foundation for implementing CQRS, sagas, and event-driven workflows without reinventing fundamental concepts. By leveraging these patterns, developers can focus on solving business problems rather than wrestling with infrastructure complexities.

For teams looking to adopt these patterns, python-cqrs offers a lightweight, Python-native solution that integrates seamlessly with existing codebases. Whether you’re building financial systems, e-commerce platforms, or real-time data pipelines, this approach ensures consistency, scalability, and resilience in the face of failures.