Python CQRS

Event-Driven Architecture Framework for Distributed Systems

🐙 Star if cool ⭐ ✨ ✨

📦 PyPI 📊 Downloads

Documentation for python-cqrs 5.x

You are viewing 5.x docs (dataclass defaults, optional Pydantic/SQLAlchemy, optional SQLModel storage since 5.1, new package layout, SagaMediator.execute).

Migrating from 4.x? Follow the 4→5 migration guide. Use the version selector for frozen 4.0 docs.

---

Core Features

• Bootstrap

  Quick project setup and configuration with automatic DI container setup.

  Read More

• Request Handlers

  Handle commands and queries with full type safety and async support.

  Read More

• Saga Pattern

  Orchestrated Saga for distributed transactions with automatic compensation.

  Read More

• Event Handling

  Process domain events with parallel processing and runtime execution.

  Read More

• Transaction Outbox

  Guaranteed event delivery with at-least-once semantics.

  Read More

• Chain of Responsibility

  Sequential request processing with flexible handler chaining.

  Read More

• Streaming

  Incremental processing with real-time progress updates via SSE.

  Read More

• Integrations

  FastAPI and FastStream integrations out of the box.

  Read More

• Mermaid Diagrams

  Visualize architecture patterns and flows with interactive Mermaid diagrams.

  Read More

---

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

Requirements

Python 3.10+ (tested on 3.10–3.13)

---

Quick Start

import di
import cqrs
from cqrs.bootstrap import requests as 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.