Python CQRS

Event-Driven Architecture Framework for Distributed Systems

Python 3.10+ · Full documentation: mkdocs.python-cqrs.dev

Warning

Breaking Changes in v5.0.0

Starting with version 5.0.0, Pydantic support will become optional. The default implementations of Request, Response, DomainEvent, and NotificationEvent will be migrated to dataclasses-based implementations.

Table of Contents

• Overview
• Installation
• Quick Start
• Request and Response Types
• Request Handlers
• Mapping
• DI container
• Bootstrap
• Saga Pattern
• Producing Notification Events
• Kafka broker
• Transactional Outbox
• Producing Events from Outbox to Kafka
• Transaction log tailing
• Event Handlers
• Integration with presentation layers
• Protobuf messaging
• Contributing
• Changelog
• License

Overview

An event-driven framework for building distributed systems in Python. It centers on CQRS (Command Query Responsibility Segregation) and extends into messaging, sagas, and reliable event delivery — so you can separate read and write flows, react to events from the bus, run distributed transactions with compensation, and publish events via Transaction Outbox. The result is clearer structure, better scalability, and easier evolution of the application.

This package is a fork of the diator project (documentation) with several enhancements, ordered by importance:

Core framework

1. Redesigned the event and request mapping mechanism to handlers;
2. EventMediator for handling Notification and ECST events coming from the bus;
3. bootstrap for easy setup;
4. Transaction Outbox, ensuring that Notification and ECST events are sent to the broker;
5. Orchestrated Saga pattern for distributed transactions with automatic compensation and recovery;
6. StreamingRequestMediator and StreamingRequestHandler for streaming requests with real-time progress updates;
7. Chain of Responsibility with CORRequestHandler for processing requests through multiple handlers in sequence;
8. Parallel event processing with configurable concurrency limits.

Also

• Typing: Pydantic v2.* and IRequest/IResponse interfaces — use Pydantic-based, dataclass-based, or custom Request/Response implementations.
• Broker: Kafka via aiokafka.
• Integration: Ready for integration with FastAPI and FastStream.
• Documentation: Built-in Mermaid diagram generation (Sequence and Class diagrams).
• Protobuf: Interface-level support for converting Notification events to Protobuf and back.

Installation

Python 3.10+ is required.

pip install python-cqrs

Optional dependencies (see pyproject.toml for full list):

pip install python-cqrs[kafka] # Kafka broker (aiokafka)
pip install python-cqrs[examples] # FastAPI, FastStream, uvicorn, etc.
pip install python-cqrs[aiobreaker] # Circuit breaker for saga fallbacks

Quick Start

Define a command, a handler, bind them, and run via the mediator:

importdiimportcqrsfromcqrs.requestsimportbootstrapclassCreateOrderCommand(cqrs.Request):
order_id: stramount: floatclassCreateOrderHandler(cqrs.RequestHandler[CreateOrderCommand, None]):
asyncdefhandle(self, request: CreateOrderCommand) ->None:
print(f"Order {request.order_id}, amount {request.amount}")
defcommands_mapper(mapper: cqrs.RequestMap) ->None:
mapper.bind(CreateOrderCommand, CreateOrderHandler)
container=di.Container()
mediator=bootstrap.bootstrap(di_container=container, commands_mapper=commands_mapper)
awaitmediator.send(CreateOrderCommand(order_id="ord-1", amount=99.99))

For full setup with DI, events, and outbox, see the documentation and the examples directory.

Request and Response Types

The library supports both Pydantic-based (PydanticRequest/PydanticResponse, aliased as Request/Response) and Dataclass-based (DCRequest/DCResponse) implementations. You can also implement custom classes by implementing the IRequest/IResponse interfaces directly.

importdataclasses# Pydantic-based (default)classCreateUserCommand(cqrs.Request):
username: stremail: strclassUserResponse(cqrs.Response):
user_id: strusername: str# Dataclass-based@dataclasses.dataclassclassCreateProductCommand(cqrs.DCRequest):
name: strprice: float@dataclasses.dataclassclassProductResponse(cqrs.DCResponse):
product_id: strname: str# Custom implementationclassCustomRequest(cqrs.IRequest):
def__init__(self, user_id: str, action: str):
self.user_id=user_idself.action=actiondefto_dict(self) ->dict:
return {"user_id": self.user_id, "action": self.action}
@classmethoddeffrom_dict(cls, **kwargs) ->"CustomRequest":
returncls(user_id=kwargs["user_id"], action=kwargs["action"])
classCustomResponse(cqrs.IResponse):
def__init__(self, result: str, status: int):
self.result=resultself.status=statusdefto_dict(self) ->dict:
return {"result": self.result, "status": self.status}
@classmethoddeffrom_dict(cls, **kwargs) ->"CustomResponse":
returncls(result=kwargs["result"], status=kwargs["status"])

A complete example can be found in request_response_types.py

Request Handlers

Request handlers can be divided into two main types:

Command Handler

Command Handler executes the received command. The logic of the handler may include, for example, modifying the state of the domain model. As a result of executing the command, an event may be produced to the broker.

Tip

By default, the command handler does not return any result, but it is not mandatory.

fromcqrs.requests.request_handlerimportRequestHandlerfromcqrs.events.eventimportEventclassJoinMeetingCommandHandler(RequestHandler[JoinMeetingCommand, None]):
def__init__(self, meetings_api: MeetingAPIProtocol) ->None:
self._meetings_api=meetings_apiself._events: list[Event] = []
@propertydefevents(self) ->typing.List[events.Event]:
returnself._eventsasyncdefhandle(self, request: JoinMeetingCommand) ->None:
awaitself._meetings_api.join_user(request.user_id, request.meeting_id)

A complete example can be found in the documentation

Query handler

Query Handler returns a representation of the requested data, for example, from the read model.

Tip

The read model can be constructed based on domain events produced by the Command Handler.

fromcqrs.requests.request_handlerimportRequestHandlerfromcqrs.events.eventimportEventclassReadMeetingQueryHandler(RequestHandler[ReadMeetingQuery, ReadMeetingQueryResult]):
def__init__(self, meetings_api: MeetingAPIProtocol) ->None:
self._meetings_api=meetings_apiself._events: list[Event] = []
@propertydefevents(self) ->typing.List[events.Event]:
returnself._eventsasyncdefhandle(self, request: ReadMeetingQuery) ->ReadMeetingQueryResult:
link=awaitself._meetings_api.get_link(request.meeting_id)
returnReadMeetingQueryResult(link=link, meeting_id=request.meeting_id)

A complete example can be found in the documentation

Streaming Request Handler

Streaming Request Handler processes requests incrementally and yields results as they become available. This is particularly useful for processing large batches of items, file uploads, or any operation that benefits from real-time progress updates.

StreamingRequestHandler works with StreamingRequestMediator that streams results to clients in real-time.

importtypingfromcqrs.requests.request_handlerimportStreamingRequestHandlerfromcqrs.events.eventimportEventclassProcessFilesCommandHandler(StreamingRequestHandler[ProcessFilesCommand, FileProcessedResult]):
def__init__(self):
self._events: list[Event] = []
@propertydefevents(self) ->list[Event]:
returnself._events.copy()
defclear_events(self) ->None:
self._events.clear()
asyncdefhandle(self, request: ProcessFilesCommand) ->typing.AsyncIterator[FileProcessedResult]:
forfile_idinrequest.file_ids:
# Process fileresult=FileProcessedResult(file_id=file_id, status="completed", ...)
# Emit eventsself._events.append(FileProcessedEvent(file_id=file_id, ...))
yieldresult

A complete example can be found in the documentation

Chain of Responsibility Request Handler

Chain of Responsibility Request Handler implements the chain of responsibility pattern, allowing multiple handlers to process a request in sequence until one successfully handles it. This pattern is particularly useful when you have multiple processing strategies or need to implement fallback mechanisms.

Each handler in the chain decides whether to process the request or pass it to the next handler. The chain stops when a handler successfully processes the request or when all handlers have been exhausted.

importtypingfromcqrs.requests.cor_request_handlerimportCORRequestHandlerfromcqrs.events.eventimportEventclassCreditCardPaymentHandler(CORRequestHandler[ProcessPaymentCommand, PaymentResult]):
def__init__(self, payment_service: PaymentServiceProtocol) ->None:
self._payment_service=payment_serviceself._events: typing.List[Event] = []
@propertydefevents(self) ->typing.List[Event]:
returnself._eventsasyncdefhandle(self, request: ProcessPaymentCommand) ->PaymentResult|None:
ifrequest.payment_method=="credit_card":
# Process credit card paymentresult=awaitself._payment_service.process_credit_card(request)
self._events.append(PaymentProcessedEvent(...))
returnPaymentResult(success=True, transaction_id=result.id)
# Pass to next handlerreturnawaitself.next(request)
classPayPalPaymentHandler(CORRequestHandler[ProcessPaymentCommand, PaymentResult]):
def__init__(self, paypal_service: PayPalServiceProtocol) ->None:
self._paypal_service=paypal_serviceself._events: typing.List[Event] = []
@propertydefevents(self) ->typing.List[Event]:
returnself._eventsasyncdefhandle(self, request: ProcessPaymentCommand) ->PaymentResult|None:
ifrequest.payment_method=="paypal":
# Process PayPal paymentresult=awaitself._paypal_service.process_payment(request)
returnPaymentResult(success=True, transaction_id=result.id)
# Pass to next handlerreturnawaitself.next(request)
# Chain registrationdefpayment_mapper(mapper: cqrs.RequestMap) ->None:
mapper.bind(ProcessPaymentCommand, [
CreditCardPaymentHandler,
PayPalPaymentHandler,
DefaultPaymentHandler# Fallback handler
])

A complete example can be found in the documentation

Mermaid Diagram Generation

The package includes built-in support for generating Mermaid diagrams from Chain of Responsibility handler chains.

fromcqrs.requests.mermaidimportCoRMermaid# Create Mermaid generator from handler chainhandlers= [CreditCardHandler, PayPalHandler, DefaultHandler]
generator=CoRMermaid(handlers)
# Generate Sequence diagram showing execution flowsequence_diagram=generator.sequence()
# Generate Class diagram showing type structureclass_diagram=generator.class_diagram()

Complete example: CoR Mermaid Diagrams

Mapping

To bind commands, queries and events with specific handlers, you can use the registries EventMap, RequestMap, and SagaMap.

Commands, queries and events:

fromcqrsimportrequests, eventsfromappimportcommands, command_handlersfromappimportqueries, query_handlersfromappimporteventsasevent_models, event_handlersdefinit_commands(mapper: requests.RequestMap) ->None:
mapper.bind(commands.JoinMeetingCommand, command_handlers.JoinMeetingCommandHandler)
definit_queries(mapper: requests.RequestMap) ->None:
mapper.bind(queries.ReadMeetingQuery, query_handlers.ReadMeetingQueryHandler)
definit_events(mapper: events.EventMap) ->None:
mapper.bind(events.NotificationEvent[event_models.NotificationMeetingRoomClosed], event_handlers.MeetingRoomClosedNotificationHandler)
mapper.bind(events.NotificationEvent[event_models.ECSTMeetingRoomClosed], event_handlers.UpdateMeetingRoomReadModelHandler)

Chain of Responsibility — bind a list of handlers (the first one that can handle the request processes it, otherwise the request is passed to the next):

defpayment_mapper(mapper: cqrs.RequestMap) ->None:
mapper.bind(
ProcessPaymentCommand,
[
CreditCardPaymentHandler,
PayPalPaymentHandler,
DefaultPaymentHandler, # Fallback
],
)

Streaming handler — bind a command to a StreamingRequestHandler (results are yielded as they become available):

defcommands_mapper(mapper: cqrs.RequestMap) ->None:
mapper.bind(ProcessOrdersCommand, ProcessOrdersCommandHandler) # StreamingRequestHandler

Saga (including with fallback) — bind the saga context type to the saga class in SagaMap:

defsaga_mapper(mapper: cqrs.SagaMap) ->None:
mapper.bind(OrderContext, OrderSaga)
mapper.bind(OrderContext, OrderSagaWithFallback)

DI container

Use the following example to set up dependency injection in your command, query and event handlers. This will make dependency management simpler.

The package supports two DI container libraries:

di library

importdi
...
defsetup_di() ->di.Container:
""" Binds implementations to dependencies """container=di.Container()
container.bind(
di.bind_by_type(
dependent.Dependent(cqrs.SqlAlchemyOutboxedEventRepository, scope="request"),
cqrs.OutboxedEventRepository
)
)
container.bind(
di.bind_by_type(
dependent.Dependent(MeetingAPIImplementaion, scope="request"),
MeetingAPIProtocol
)
)
returncontainer

A complete example can be found in the documentation

dependency-injector library

The package also supports dependency-injector library. You can use DependencyInjectorCQRSContainer adapter to integrate dependency-injector containers with python-cqrs.

fromdependency_injectorimportcontainers, providersfromcqrs.container.dependency_injectorimportDependencyInjectorCQRSContainerclassApplicationContainer(containers.DeclarativeContainer):
# Define your providersservice=providers.Factory(ServiceImplementation)
# Create CQRS container adaptercqrs_container=DependencyInjectorCQRSContainer(ApplicationContainer())
# Use with bootstrapmediator=bootstrap.bootstrap(
di_container=cqrs_container,
commands_mapper=commands_mapper,
...
)

Complete examples can be found in:

• Simple example
• Practical example with FastAPI

Bootstrap

The python-cqrs package implements a set of bootstrap utilities designed to simplify the initial configuration of an application.

importfunctoolsfromcqrs.eventsimportbootstrapasevent_bootstrapfromcqrs.requestsimportbootstrapasrequest_bootstrapfromappimportdependencies, mapping, orm@functools.lru_cachedefmediator_factory():
returnrequest_bootstrap.bootstrap(
di_container=dependencies.setup_di(),
commands_mapper=mapping.init_commands,
queries_mapper=mapping.init_queries,
domain_events_mapper=mapping.init_events,
on_startup=[orm.init_store_event_mapper],
)
@functools.lru_cachedefevent_mediator_factory():
returnevent_bootstrap.bootstrap(
di_container=dependencies.setup_di(),
events_mapper=mapping.init_events,
on_startup=[orm.init_store_event_mapper],
)
@functools.lru_cachedefsaga_mediator_factory():
returnsaga_bootstrap.bootstrap(
di_container=dependencies.setup_di(),
sagas_mapper=mapping.init_sagas,
domain_events_mapper=mapping.init_events,
saga_storage=MemorySagaStorage(),
)

Saga Pattern

The package implements the Orchestrated Saga pattern for managing distributed transactions across multiple services or operations. Sagas enable eventual consistency by executing a series of steps where each step can be compensated if a subsequent step fails.

Key Features

• SagaStorage: Persists saga state and execution history, enabling recovery of interrupted sagas
• SagaLog: Tracks all step executions (act/compensate) with status and timestamps
• Recovery Mechanism: Automatically recovers interrupted sagas from storage, ensuring eventual consistency
• Automatic Compensation: If any step fails, all previously completed steps are automatically compensated in reverse order
• Fallback Pattern: Define alternative steps to execute when primary steps fail, with optional Circuit Breaker protection
• Mermaid Diagram Generation: Generate Sequence and Class diagrams for documentation and visualization

Example

importdataclassesimportuuidfromcqrs.saga.modelsimportSagaContextfromcqrs.saga.sagaimportSagafromcqrs.saga.stepimportSagaStepHandler@dataclasses.dataclassclassOrderContext(SagaContext):
order_id: struser_id: stritems: list[str]
total_amount: floatinventory_reservation_id: str|None=Nonepayment_id: str|None=None# Define saga class with stepsclassOrderSaga(Saga[OrderContext]):
steps= [
ReserveInventoryStep,
ProcessPaymentStep,
]
# Execute saga via mediatorcontext=OrderContext(order_id="123", user_id="user_1", items=["item_1"], total_amount=100.0)
saga_id=uuid.uuid4()
asyncforstep_resultinmediator.stream(context, saga_id=saga_id):
print(f"Step completed: {step_result.step_type.__name__}")
# If any step fails, compensation happens automatically

Fallback Pattern with Circuit Breaker

The saga pattern supports fallback steps that execute automatically when primary steps fail. You can also integrate Circuit Breaker protection to prevent cascading failures:

fromcqrs.saga.fallbackimportFallbackfromcqrs.adapters.circuit_breakerimportAioBreakerAdapterfromcqrs.responseimportResponsefromcqrs.saga.stepimportSagaStepHandler, SagaStepResultclassReserveInventoryResponse(Response):
reservation_id: strclassPrimaryStep(SagaStepHandler[OrderContext, ReserveInventoryResponse]):
asyncdefact(self, context: OrderContext) ->SagaStepResult[OrderContext, ReserveInventoryResponse]:
# Primary step that may failraiseRuntimeError("Service unavailable")
classFallbackStep(SagaStepHandler[OrderContext, ReserveInventoryResponse]):
asyncdefact(self, context: OrderContext) ->SagaStepResult[OrderContext, ReserveInventoryResponse]:
# Alternative step that executes when primary failsreservation_id=f"fallback_reservation_{context.order_id}"context.reservation_id=reservation_idreturnself._generate_step_result(ReserveInventoryResponse(reservation_id=reservation_id))
# Define saga with fallback and circuit breakerclassOrderSagaWithFallback(Saga[OrderContext]):
steps= [
Fallback(
step=PrimaryStep,
fallback=FallbackStep,
circuit_breaker=AioBreakerAdapter(
fail_max=2, # Circuit opens after 2 failurestimeout_duration=60, # Wait 60 seconds before retry
),
),
]
# Optional: Using Redis for distributed circuit breaker state# import redis# from aiobreaker.storage.redis import CircuitRedisStorage## def redis_storage_factory(name: str):# client = redis.from_url("redis://localhost:6379", decode_responses=False)# return CircuitRedisStorage(state="closed", redis_object=client, namespace=name)## AioBreakerAdapter(..., storage_factory=redis_storage_factory)

When the primary step fails, the fallback step executes automatically. The Circuit Breaker opens after the configured failure threshold, preventing unnecessary load on failing services by failing fast.

The saga state and step history are persisted to SagaStorage. The SagaLog maintains a complete audit trail of all step executions (both act and compensate operations) with timestamps and status information. This enables the recovery mechanism to restore saga state and ensure eventual consistency even after system failures.

If a saga is interrupted (e.g., due to a crash), you can recover it using the recovery mechanism:

fromcqrs.saga.recoveryimportrecover_saga# Get saga instance from mediator's saga map (or keep reference to saga class)saga=OrderSaga()
# Recover interrupted saga - will resume from last completed step# or continue compensation if saga was in compensating stateawaitrecover_saga(
saga=saga,
saga_id=saga_id,
context_builder=OrderContext,
container=di_container, # Same container used in bootstrapstorage=storage,
)
# Access execution history (SagaLog) for monitoring and debugginghistory=awaitstorage.get_step_history(saga_id)
forentryinhistory:
print(f"{entry.timestamp}: {entry.step_name} - {entry.action} - {entry.status}")

The recovery mechanism ensures eventual consistency by:

• Loading the last known saga state from SagaStorage
• Checking the SagaLog to determine which steps were completed
• Resuming execution from the last completed step, or continuing compensation if the saga was in a compensating state
• Preventing duplicate execution of already completed steps

Mermaid Diagram Generation

The package includes built-in support for generating Mermaid diagrams from Saga instances.

fromcqrs.saga.mermaidimportSagaMermaid# Create Mermaid generator from saga classsaga=OrderSaga()
generator=SagaMermaid(saga)
# Generate Sequence diagram showing execution flowsequence_diagram=generator.sequence()
# Generate Class diagram showing type structureclass_diagram=generator.class_diagram()

Complete example: Saga Mermaid Diagrams

Producing Notification Events

During the handling of a command, cqrs.NotificationEvent events may be generated and then sent to the broker.

classJoinMeetingCommandHandler(cqrs.RequestHandler[JoinMeetingCommand, None]):
def__init__(self):
self._events= []
@propertydefevents(self):
returnself._eventsasyncdefhandle(self, request: JoinMeetingCommand) ->None:
print(f"User {request.user_id} joined meeting {request.meeting_id}")
self._events.append(
cqrs.NotificationEvent[UserJoinedNotificationPayload](
event_name="UserJoined",
topic="user_notification_events",
payload=UserJoinedNotificationPayload(
user_id=request.user_id,
meeting_id=request.meeting_id,
),
)
)
self._events.append(
cqrs.NotificationEvent[UserJoinedECSTPayload](
event_name="UserJoined",
topic="user_ecst_events",
payload=UserJoinedECSTPayload(
user_id=request.user_id,
meeting_id=request.meeting_id,
),
)
)

A complete example can be found in the documentation

After processing the command/request, if there are any Notification/ECST events, the EventEmitter is invoked to produce the events via the message broker.

Warning

It is important to note that producing events using the events property parameter does not guarantee message delivery to the broker. In the event of broker unavailability or an exception occurring during message formation or sending, the message may be lost. This issue can potentially be addressed by configuring retry attempts for sending messages to the broker, but we recommend using the Transaction Outbox pattern, which is implemented in the current version of the python-cqrs package for this purpose.

Kafka broker

fromcqrs.adaptersimportkafkaaskafka_adapterfromcqrs.message_brokersimportkafkaaskafka_brokerproducer=kafka_adapter.kafka_producer_factory(
dsn="localhost:9092",
topics=["test.topic1", "test.topic2"],
)
broker=kafka_broker.KafkaMessageBroker(producer)
awaitbroker.send_message(...)

Transactional Outbox

The package implements the Transactional Outbox pattern, which ensures that messages are produced to the broker according to the at-least-once semantics.

classJoinMeetingCommandHandler(cqrs.RequestHandler[JoinMeetingCommand, None]):
def__init__(self, outbox: cqrs.OutboxedEventRepository):
self.outbox=outbox@propertydefevents(self):
return []
asyncdefhandle(self, request: JoinMeetingCommand) ->None:
print(f"User {request.user_id} joined meeting {request.meeting_id}")
# Outbox repository is bound to a session (e.g. via DI request scope).# add() takes only the event; commit() persists the outbox and your changes.self.outbox.add(
cqrs.NotificationEvent[UserJoinedNotificationPayload](
event_name="UserJoined",
topic="user_notification_events",
payload=UserJoinedNotificationPayload(
user_id=request.user_id,
meeting_id=request.meeting_id,
),
),
)
self.outbox.add(
cqrs.NotificationEvent[UserJoinedECSTPayload](
event_name="UserJoined",
topic="user_ecst_events",
payload=UserJoinedECSTPayload(
user_id=request.user_id,
meeting_id=request.meeting_id,
),
),
)
awaitself.outbox.commit()

A complete example can be found in the documentation

Tip

You can specify the name of the Outbox table using the environment variable OUTBOX_SQLA_TABLE. By default, it is set to outbox.

Tip

If you use the protobuf events you should specify OutboxedEventRepository by protobuf serialize. A complete example can be found in the documentation

Producing Events from Outbox to Kafka

As an implementation of the Transactional Outbox pattern, the SqlAlchemyOutboxedEventRepository is available for use as an access repository to the Outbox storage. It can be utilized in conjunction with the KafkaMessageBroker.

importasyncioimportcqrsfromcqrs.message_brokersimportkafkafromcqrs.adaptersimportkafkaaskafka_adaptersfromcqrs.compressorsimportzlibsession_factory=async_sessionmaker(
create_async_engine(
f"mysql+asyncmy://{USER}:{PASSWORD}@{HOSTNAME}:{PORT}/{DATABASE}",
isolation_level="REPEATABLE READ",
)
)
broker=kafka.KafkaMessageBroker(
producer=kafka_adapters.kafka_producer_factory(dsn="localhost:9092"),
)
# SqlAlchemyOutboxedEventRepository expects (session, compressor), not session_factory.asyncwithsession_factory() assession:
repository=cqrs.SqlAlchemyOutboxedEventRepository(session, zlib.ZlibCompressor())
producer=cqrs.EventProducer(broker, repository)
asyncformessagesinproducer.event_batch_generator():
formessageinmessages:
awaitproducer.send_message(message)
awaitproducer.repository.commit()
awaitasyncio.sleep(10)

A complete example can be found in the documentation

Transaction log tailing

If the Outbox polling strategy does not suit your needs, I recommend exploring the Transaction Log Tailing pattern. The current version of the python-cqrs package does not support the implementation of this pattern.

Tip

However, it can be implemented using Debezium + Kafka Connect, which allows you to produce all newly created events within the Outbox storage directly to the corresponding topic in Kafka (or any other broker).

Event Handlers

Event handlers are designed to process Notification and ECST events that are consumed from the broker. To configure event handling, you need to implement a broker consumer on the side of your application. Below is an example of Kafka event consuming that can be used in the Presentation Layer.

classJoinMeetingCommandHandler(cqrs.RequestHandler[JoinMeetingCommand, None]):
def__init__(self):
self._events= []
@propertydefevents(self):
returnself._eventsasyncdefhandle(self, request: JoinMeetingCommand) ->None:
STORAGE[request.meeting_id].append(request.user_id)
self._events.append(
UserJoined(user_id=request.user_id, meeting_id=request.meeting_id),
)
print(f"User {request.user_id} joined meeting {request.meeting_id}")
classUserJoinedEventHandler(cqrs.EventHandler[UserJoined]):
asyncdefhandle(self, event: UserJoined) ->None:
print(f"Handle user {event.user_id} joined meeting {event.meeting_id} event")

A complete example can be found in the documentation

Parallel Event Processing

Both RequestMediator and StreamingRequestMediator support parallel processing of domain events. You can control the number of event handlers that run simultaneously using the max_concurrent_event_handlers parameter.

This feature is especially useful when:

• Multiple event handlers need to process events independently
• You want to improve performance by processing events concurrently
• You need to limit resource consumption by controlling concurrency

Configuration:

fromcqrs.requestsimportbootstrapmediator=bootstrap.bootstrap_streaming(
di_container=container,
commands_mapper=commands_mapper,
domain_events_mapper=domain_events_mapper,
message_broker=broker,
max_concurrent_event_handlers=3, # Process up to 3 events in parallelconcurrent_event_handle_enable=True, # Enable parallel processing
)

Tip

• Set max_concurrent_event_handlers to limit the number of simultaneously running event handlers
• Set concurrent_event_handle_enable=False to disable parallel processing and process events sequentially
• The default value for max_concurrent_event_handlers is 10 for StreamingRequestMediator and 1 for RequestMediator

Integration with presentation layers

The framework is ready for integration with FastAPI and FastStream.

Tip

I recommend reading the useful paper Onion Architecture Used in Software Development. Separating user interaction and use-cases into Application and Presentation layers is a good practice. This can improve the Testability, Maintainability, Scalability of the application. It also provides benefits such as Separation of Concerns.

FastAPI requests handling

If your application uses FastAPI (or any other asynchronous framework for creating APIs). In this case you can use python-cqrs to route requests to the appropriate handlers implementing specific use-cases.

importfastapiimportpydanticfromappimportdependencies, commandsrouter=fastapi.APIRouter(prefix="/meetings")
@router.put("/{meeting_id}/{user_id}", status_code=status.HTTP_200_OK)asyncdefjoin_metting(
meeting_id: pydantic.PositiveInt,
user_id: typing.Text,
mediator: cqrs.RequestMediator=fastapi.Depends(dependencies.mediator_factory),
):
awaitmediator.send(commands.JoinMeetingCommand(meeting_id=meeting_id, user_id=user_id))
return {"result": "ok"}

A complete example can be found in the documentation

Kafka events consuming

If you build interaction by events over broker like Kafka, you can to implement an event consumer on your application's side, which will call the appropriate handler for each event. An example of handling events from Kafka is provided below.

importcqrsimportpydanticimportfaststreamfromfaststreamimportkafkabroker=kafka.KafkaBroker(bootstrap_servers=["localhost:9092"])
app=faststream.FastStream(broker)
classHelloWorldPayload(pydantic.BaseModel):
hello: str=pydantic.Field(default="Hello")
world: str=pydantic.Field(default="World")
classHelloWorldECSTEventHandler(cqrs.EventHandler[cqrs.NotificationEvent[HelloWorldPayload]]):
asyncdefhandle(self, event: cqrs.NotificationEvent[HelloWorldPayload]) ->None:
print(f"{event.payload.hello}{event.payload.world}") # type: ignore@broker.subscriber("hello_world",group_id="examples",auto_commit=False,value_deserializer=value_deserializer,decoder=decoder,)asyncdefhello_world_event_handler(
body: cqrs.NotificationEvent[HelloWorldPayload] |None,
msg: kafka.KafkaMessage,
mediator: cqrs.EventMediator=faststream.Depends(mediator_factory),
):
ifbodyisnotNone:
awaitmediator.send(body)
awaitmsg.ack()

A complete example can be found in the documentation

FastAPI SSE Streaming

StreamingRequestMediator is ready and designed for use with Server-Sent Events (SSE) in FastAPI applications. This allows you to stream results to clients in real-time as they are processed.

Example FastAPI endpoint with SSE:

importfastapiimportjsonfromcqrs.requestsimportbootstrapdefstreaming_mediator_factory() ->cqrs.StreamingRequestMediator:
returnbootstrap.bootstrap_streaming(
di_container=container,
commands_mapper=commands_mapper,
domain_events_mapper=domain_events_mapper,
message_broker=broker,
max_concurrent_event_handlers=3,
concurrent_event_handle_enable=True,
)
@app.post("/process-files")asyncdefprocess_files_stream(
command: ProcessFilesCommand,
mediator: cqrs.StreamingRequestMediator=fastapi.Depends(streaming_mediator_factory),
) ->fastapi.responses.StreamingResponse:
asyncdefgenerate_sse():
yieldf"data: {json.dumps({'type': 'start', 'message': 'Processing...'})}\\n\\n"asyncforresultinmediator.stream(command):
sse_data= {
"type": "progress",
"data": result.to_dict(),
}
yieldf"data: {json.dumps(sse_data)}\\n\\n"yieldf"data: {json.dumps({'type': 'complete'})}\\n\\n"returnfastapi.responses.StreamingResponse(
generate_sse(),
media_type="text/event-stream",
)

A complete example can be found in the documentation

Protobuf messaging

The python-cqrs package supports integration with protobuf. Notification events can be serialized to Protobuf and back: implement the proto() method (returns a protobuf message) and the class method from_proto() (creates an event instance from proto) on your event class.

Example (assuming generated user_joined_pb2 from your .proto with fields event_id, event_timestamp, event_name, payload):

importuuidfromdatetimeimportdatetimeimportcqrsfromapp.generatedimportuser_joined_pb2# generated from .protoclassUserJoinedPayload(cqrs.Response):
user_id: strmeeting_id: strclassUserJoinedNotificationEvent(cqrs.NotificationEvent[UserJoinedPayload]):
"""Event with Protobuf serialization support."""event_name: str="UserJoined"defproto(self):
msg=user_joined_pb2.UserJoinedNotification()
msg.event_id=str(self.event_id)
msg.event_timestamp=self.event_timestamp.isoformat()
msg.event_name=self.event_namemsg.payload.user_id=self.payload.user_idmsg.payload.meeting_id=self.payload.meeting_idreturnmsg@classmethoddeffrom_proto(cls, proto_msg):
returncls(
event_id=uuid.UUID(proto_msg.event_id),
event_timestamp=datetime.fromisoformat(proto_msg.event_timestamp),
event_name=proto_msg.event_name,
topic="user_notification_events",
payload=UserJoinedPayload(
user_id=proto_msg.payload.user_id,
meeting_id=proto_msg.payload.meeting_id,
),
)

Contributing

Contributions are welcome. To develop locally:

1. Clone the repository and create a virtual environment.
2. Install dev dependencies: pip install -e ".[dev]".
3. Run tests: pytest.
4. Install pre-commit and run hooks: pre-commit install && pre-commit run --all-files.

The project uses ruff for linting and pyright for type checking.

Changelog

Release notes and migration guides are published on GitHub Releases.

License

This project is licensed under the MIT License — see the LICENSE file for details.