pypatterns/python-cqrs

This system is a Python CQRS (Command Query Responsibility Segregation) library that provides a structured framework for building event-driven applications. It facilitates command and query handling through mediators, supports complex distributed transactions via a saga pattern with compensation and recovery, and ensures reliable event delivery using an outbox pattern. The library integrates with message brokers like Kafka and AMQP, offers resilience features such as circuit breakers and fallbacks, and allows for flexible dependency injection and middleware customization.

17 changes.

This period established the foundational architecture of the python-cqrs library, introducing core components for command and query handling, event mediation, and saga orchestration. It implemented essential infrastructure including message broker adapters for AMQP and Kafka, an outbox pattern for reliable event persistence, and dependency injection containers. The work also included comprehensive test suites and example applications to validate the framework's capabilities.

4 changes.

This period focused on implementing a comprehensive Saga pattern for managing distributed transactions, featuring compensation, fallbacks, and recovery mechanisms. The work included building a configurable storage layer with both in-memory and persistent backends, alongside extensive unit testing and performance benchmarking to validate the new CQRS components.

Users can now resolve dependencies in CQRS handlers using two supported backends: the \dependency-injector\ library and the \di\ library. The \DependencyInjectorCQRSContainer\ adapter automatically discovers providers from an external \dependency-injector\ container, supporting both exact type matches and inheritance-based resolution for interfaces. The \DIContainer\ adapter integrates with the \di\ library, using its \AsyncExecutor\ to resolve dependencies within a request-scoped context. Both adapters implement the \Container\ protocol, allowing seamless switching between DI strategies while maintaining a consistent API for handler dependency resolution.

src/cqrs/container · high confidence

The examples/proto directory now includes the source definition (user\_joined.proto) and the generated Python code (user\_joined\_pb2.py) for the UserJoinedECST and UserJoinedNotification message structures. These files provide the schema and serialization logic for events containing user\_id, meeting\_id, event\_id, event\_timestamp, and event\_name, enabling developers to use these specific Protobuf types in their examples.

examples/proto · high confidence

A new shell script, \scripts/update\_run.sh\, has been added to support running the Kafka cluster in KRaft mode (without Zookeeper). The script applies Docker workarounds by removing Zookeeper connection checks and ignoring Zookeeper readiness probes, and it formats the storage directory with a new cluster ID to satisfy KRaft requirements.

scripts · high confidence

This change introduces the initial implementation of the CQRS adapter layer, providing concrete integrations for message brokers and resilience patterns. It adds AMQP support via \aio\_pika\ (including publisher and consumer factories with connection pooling) and Kafka support via \aiokafka\ (featuring configurable security protocols, SASL mechanisms, SSL context support, and automatic retry logic). Additionally, it implements a circuit breaker adapter using \aiobreaker\ to handle failures in saga steps and request/event handling, along with the underlying protocol definitions that decouple these interfaces from specific implementation details.

src/cqrs/adapters · high confidence

Establishes the foundational development environment for the python-cqrs library by adding pre-commit hooks (Ruff, Pyright, Vermin), configuration files for linting and type checking, and documentation for local setup and contribution. Includes Docker Compose files for local Kafka, MySQL, PostgreSQL, and Redis services, a .gitignore update, and a py.typed marker to enable static type checking for consumers.

(repo-wide) · high confidence

This change introduces the core components of the CQRS (Command Query Responsibility Segregation) library, establishing the foundational architecture for the project. It provides a unified mediator system with distinct classes for handling requests (RequestMediator), events (EventMediator), and streaming responses (StreamingRequestMediator). The library supports flexible data modeling through Pydantic and dataclass-based events and responses, and includes robust infrastructure features such as an outbox pattern for reliable event persistence with dialect-aware SQLAlchemy types (supporting PostgreSQL, MySQL, and MariaDB), event compression via Zlib, and circuit breaker patterns for handling fallbacks in saga, request, and event processing. Additionally, it exposes utilities for generating Mermaid diagrams for sagas and command flows, and a DI container for dependency management.

src/cqrs · high confidence

The dispatcher module now provides dedicated dispatchers for requests, events, and sagas, enabling more robust command and query handling. Request and event dispatchers support fallback mechanisms with circuit breaker integration, allowing the system to switch to alternative handlers when primary ones fail. A new streaming request dispatcher supports async generator handlers for real-time response streaming, while the saga dispatcher manages multi-step transactions with configurable compensation retries and event propagation. All dispatchers integrate with the middleware chain and dependency injection container.

src/cqrs/dispatcher · high confidence

The \src/cqrs/events\ module now provides a complete event-driven architecture for the CQRS layer. Users can define events using either Pydantic models (default) or frozen dataclasses, and register handlers via an \EventMap\. The \EventMediator\ (bootstrapped via \setup\_mediator\ or \bootstrap\) coordinates event dispatch, supporting both in-process domain events and external notification events sent to a message broker. The \EventProcessor\ allows handling events sequentially or in parallel with concurrency limits. Additionally, the system supports resilient handler execution through \EventHandlerFallback\, which wraps handlers with optional circuit breakers to switch to a fallback handler on failure.

src/cqrs/events · high confidence

This change adds the core infrastructure for the outbox pattern in the CQRS module. It introduces an abstract repository interface and a concrete SQLAlchemy implementation to store notification events reliably in a database before they are published to a message broker. The implementation includes an event mapping registry to validate event types, support for optional payload compression, and a mock repository for testing. Users can now ensure reliable message delivery by persisting events in an outbox table with status tracking (new, produced, not produced) and topic filtering.

src/cqrs/outbox · high confidence

The CQRS module now includes a middleware system that allows intercepting and processing requests before they reach their handlers. This change introduces base classes for middleware chains (\Middleware\, \MiddlewareChain\) and a specific \LoggingMiddleware\ that records debug-level logs for incoming requests and outgoing responses, including masked JSON representations of the data.

src/cqrs/middlewares · high confidence

The CQRS module now exposes a standardized \MessageBroker\ protocol and \Message\ data structure, allowing users to decouple message handling from specific transport mechanisms. This change introduces concrete implementations for AMQP (via \aio\_pika\) and Kafka (via \aiokafka\), enabling messages to be published to specific topics with configurable logging levels. A \DevnullMessageBroker\ is also provided for testing or development scenarios where messages are logged but not persisted. This establishes the foundation for flexible message routing within the application's command and query side.

_src/cqrs/message\brokers · high confidence

The examples directory now includes demonstrations for the Chain of Responsibility pattern, including handler chains, Mermaid diagram generation, and request fallbacks when the chain fails. It also adds examples for event handler fallbacks with optional circuit breaker support, and shows how to integrate the CQRS framework with both the \di\ and \dependency-injector\ libraries for dependency injection.

examples · high confidence

The \src/cqrs/requests\ module now provides a complete request handling layer, introducing \RequestHandler\ and \StreamingRequestHandler\ interfaces for standard and async-streaming commands, alongside a \CORRequestHandler\ that implements the chain of responsibility pattern for sequential processing. Users can now define fallback strategies via \RequestHandlerFallback\ to handle failures or circuit-breaker states, and map requests to handlers using the new \RequestMap\ and \SagaMap\ registries. The module also includes \bootstrap.py\ for initializing the mediator with dependency injection and middlewares, and \mermaid.py\ to generate sequence and class diagrams for handler chains.

src/cqrs/requests · high confidence

The \src/cqrs/saga\ module introduces a complete Saga implementation for managing distributed transactions. It provides a \SagaMediator\ and \bootstrap\ functions to configure saga execution, including support for choreographic sagas via event handlers. Key capabilities include automatic compensation (rollback) of completed steps in reverse order upon failure, with configurable retry and exponential backoff. The system supports fallback steps via a \Fallback\ wrapper, allowing alternative logic when a primary step fails or a circuit breaker is open. Additionally, it features saga recovery to resume interrupted sagas from storage, strict backward recovery to prevent zombie states, and Mermaid diagram generation for visualizing saga flows.

src/cqrs/saga · high confidence

The saga storage subsystem now exposes a formalized interface (ISagaStorage) and concrete implementations for both in-memory (MemorySagaStorage) and persistent (SqlAlchemySagaStorage) backends. The persistent implementation introduces configurable table names for saga executions and logs via the CQRS\_SAGA\_EXECUTION\_TABLE\_NAME and CQRS\_SAGA\_LOG\_TABLE\_NAME environment variables, allowing users to customize database schema naming. The storage layer also supports optimistic locking through version tracking and provides detailed step logging for saga execution history.

src/cqrs/saga/storage · high confidence

The CQRS module now uses a new, protocol-based approach for message serialization and deserialization, decoupling the implementation from specific frameworks like Pydantic. The \JsonDeserializer\ converts JSON data into objects that implement the \Deserializable\ protocol (requiring a \from\_dict\ classmethod), while the \default\_serializer\ handles outgoing messages by calling \to\_dict()\ or \model\_dump()\ if available. This change introduces \orjson\ as the underlying JSON engine and provides a \DeserializeJsonError\ dataclass for structured error reporting during deserialization failures.

src/cqrs/deserializers · high confidence

Add benchmarks for CQRS components; Added unit tests for CQRS framework components; Added unit tests for saga orchestration, fallback, recovery, and storage; Initial test infrastructure setup; Integration test suite for CQRS, Saga, and Outbox components.

Written by watchdog.canine.dev from the codebase's own history, inside the signed delivery this page is composed from.

• CAI 53 → 60 (+7.2)
• Rubric changed (rubric-2026.08.19 → rubric-2026.09.15) — scores are not directly comparable.

• Code Health 85 → 90 (+5.4)
• Architecture 100 → 98 (-1.8)
• Maturity 63 → 63 (+0.1)
• Readiness 41 → 46 (+5.5)
• Security 47 → 65 (+18.0)
• Domain Modelling 100 → 100 (+0.0)

• Coverage not included — suite not readable by the collector
• Dependency hygiene not measured — dependency manifest found but not parsed for hygiene
• Duplicated block (10 lines × 2) (src/cqrs/requests/mermaid.py)
• Duplicated block (10 lines × 2) (src/cqrs/saga/mermaid.py)
• Duplicated block (10 lines × 2) (src/cqrs/saga/storage/sqlalchemy.py)
• Duplicated block (10 lines × 2) (src/cqrs/saga/storage/sqlalchemy.py)
• Duplicated block (11 lines × 2) (examples/saga.py)
• Duplicated block (11 lines × 2) (src/cqrs/requests/mermaid.py)
• Duplicated block (11 lines × 2) (src/cqrs/requests/mermaid.py)
• Duplicated block (11 lines × 3) (src/cqrs/saga/mermaid.py)
• Duplicated block (13 lines × 2) (examples/saga_recovery.py)
• Duplicated block (13 lines × 2) (src/cqrs/events/fallback.py)
• Duplicated block (14 lines × 2) (examples/cor_mermaid.py)
• Duplicated block (15 lines × 2) (examples/saga_recovery.py)
• Duplicated block (16 lines × 2) (examples/cor_request_handler.py)
• Duplicated block (16 lines × 3) (examples/cor_mermaid.py)
• Duplicated block (17 lines × 2) (examples/saga.py)
• Duplicated block (19 lines × 2) (examples/saga.py)
• Duplicated block (19 lines × 3) (src/cqrs/dispatcher/request.py)
• Duplicated block (20 lines × 2) (examples/saga.py)
• …and 21 more

• Dependency hygiene PARTLY measured — Python dependencies read, no exact pin to grade for currency
• Documentation: no installation or build instructions (README.md)
• Documentation: no usage examples (README.md)
• Duplicated block (10 lines × 4) (examples/saga_recovery.py)
• Duplicated block (10 lines × 4) (src/cqrs/requests/mermaid.py)
• Duplicated block (10–11 lines × 2) (examples/saga.py)
• Duplicated block (10–12 lines × 2) (src/cqrs/saga/storage/sqlalchemy.py)
• Duplicated block (10–13 lines × 3) (examples/saga.py)
• Duplicated block (10–13 lines × 3) (src/cqrs/saga/mermaid.py)
• Duplicated block (114 lines × 2) (src/cqrs/requests/request.py)
• Duplicated block (11–13 lines × 3) (examples/cor_request_handler.py)
• Duplicated block (12 lines × 2) (src/cqrs/events/fallback.py)
• Duplicated block (12 lines × 2) (src/cqrs/requests/mermaid.py)
• Duplicated block (12 lines × 2) (src/cqrs/saga/mermaid.py)
• Duplicated block (12 lines × 3) (examples/saga_recovery.py)
• Duplicated block (14 lines × 2) (examples/saga_recovery.py)
• Duplicated block (14–15 lines × 2) (examples/saga_recovery.py)
• Duplicated block (15 lines × 4) (src/cqrs/requests/mermaid.py)
• Duplicated block (15–16 lines × 3) (src/cqrs/saga/mermaid.py)
• Duplicated block (17–22 lines × 3) (examples/saga.py)
• …and 45 more

• 10 commits — 7 feature/other, 3 fixes

• (root) — 8 commits
• (repo) — 1 commit
• src/cqrs — 1 commit

• fix: Fix Outbox PostgreSQL schema and add dialect-aware SQLAlchemy types.
• fix: Fix Outbox PostgreSQL support with dialect-aware SQLAlchemy types (#87)
• fix: Revert "Handle domain event handlers concurrently with exception isolation"
• change: Address CodeRabbit nits on README tips and Alembic test cleanup.
• change: Bump aiohttp from 3.14.1 to 3.14.3 (#78)
• change: Bump cryptography from 48.0.1 to 50.0.0 (#77)
• change: Handle domain event handlers concurrently with exception isolation
• change: Handle domain event handlers concurrently with exception isolation (#76)
• change: docs: add CONTRIBUTING.md with local setup, tooling, and PR guidelines (#83)
• change: docs: when not to use python-cqrs; examples: FastAPI + Outbox sample (#86)

Written by watchdog.canine.dev from the codebase's own history, inside the signed delivery this page is composed from.

pypatterns/python-cqrs was measured the same way every project in this corpus was: the same rubric, at a pinned commit, with the result published in full. Point a surveyor at a repository you know and see whether you agree with it.