Python CQRS

Event-Driven Architecture Framework for Distributed Systems

🐙 Star if cool ⭐ ✨ ✨

📦 PyPI 📊 Downloads

Core Features¶

What is it?¶

Python CQRS is a framework for implementing the CQRS (Command Query Responsibility Segregation) pattern in Python applications. It helps separate read and write operations, improving scalability, performance, and code maintainability.

Key Highlights:

• Performance — Separation of commands and queries, parallel event processing
• Reliability — Transaction Outbox for guaranteed event delivery, Saga with compensation support and eventual consistency
• Flexible Types — Easy integration with any type: Pydantic, dataclasses, msgspec, attrs, TypedDict and more
• Ready Integrations — FastAPI and FastStream out of the box
• Simple Setup — Bootstrap for quick configuration
• Proven Patterns — CQRS, Saga, Outbox and more to keep services decoupled and maintainable

Project status¶

Group	Badges
Python version & PyPI
Downloads
Quality & CI
Documentation & community

Installation¶

Install Python CQRS using pip or uv:

Using pip:

pip install python-cqrs

Using uv:

uv pip install python-cqrs

Quick Start¶

import di
import cqrs
from cqrs.requests import bootstrap

# Define command, response and handler
class CreateUserCommand(cqrs.Request):
    email: str
    name: str

class CreateUserResponse(cqrs.Response):
    user_id: str

class CreateUserHandler(cqrs.RequestHandler[CreateUserCommand, CreateUserResponse]):
    async def handle(self, request: CreateUserCommand) -> CreateUserResponse:
        # Your business logic here
        user_id = f"user_{request.email}"
        return CreateUserResponse(user_id=user_id)

# Bootstrap and use
mediator = bootstrap.bootstrap(
    di_container=di.Container(),
    commands_mapper=lambda m: m.bind(CreateUserCommand, CreateUserHandler),
)

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

See Bootstrap for detailed setup instructions.