python-cq
python-cq is an async-first Python library for organizing code around CQRS. It separates reads (queries), writes (commands), and notifications (events) into dedicated message buses, and lets you plug in any dependency injection framework behind a small protocol.
What is CQRS?
CQRS (Command Query Responsibility Segregation) splits read operations from write operations. Each operation has a single, well-defined responsibility, which:
- clarifies intent: a
CreateUserCommanddoes one thing, and its name says so; - keeps handlers small: one message, one handler, easy to test in isolation;
- makes side effects explicit: events fan out to subscribers without coupling the producer to them.
CQRS is often discussed alongside distributed systems and Event Sourcing, but the pattern is just as useful in a local or monolithic application. The boundaries it draws are valuable on their own.
Three message types
| Type | Intent | Handlers | Returns |
|---|---|---|---|
Command |
Change the state of the system | Exactly one | The handler's return value |
Query |
Read state without side effects | Exactly one | The handler's return value |
Event |
Notify that something has happened | Zero, one, or many | Nothing |
A Command is allowed to return a value (for convenience, typically an id or a result object), but that does not mean it should be used as a query. Keep intent clear.
Installation
Requires Python 3.12 or higher.
With the default DI backend (python-injection, recommended):
pip install "python-cq[injection]"
Without dependency injection (you will need to implement a DIAdapter):
pip install python-cq
Quickstart
import asyncio
from cq import CommandBus, command_handler
from dataclasses import dataclass
from injection import inject
@dataclass
class CreateUserCommand:
name: str
email: str
@command_handler
class CreateUserHandler:
async def handle(self, command: CreateUserCommand) -> int:
# ... persist the user, return its id
return 42
@inject
async def main(bus: CommandBus[int]) -> None:
command = CreateUserCommand(name="Ada", email="ada@example.com")
user_id = await bus.dispatch(command)
print(f"Created user {user_id}")
asyncio.run(main())
The decorator registers the handler against the type of its first handle parameter. The bus is resolved by the DI container and dispatched to that handler.
Prerequisites
Familiarity with the following helps you get the most out of python-cq:
- CQRS, in particular the distinction between Commands, Queries, and Events.
- Domain Driven Design (DDD), particularly aggregates and bounded contexts, which complement CQRS well.
1 maintainer
Release files for python-cq 0.23.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| python_cq-0.23.0.tar.gz | 16.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| python_cq-0.23.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 38.6 kB
Release files / python_cq-0.23.0.tar.gz
| Download URL | python_cq-0.23.0.tar.gz |
|---|---|
| Size | 16.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
7e28ad8d6b8dc75881d88cc00c3ab2d4ae8f57906a14aba1863f9e0b2acac538
|
|
BLAKE2b-256 checksum How to use checksums |
4911e54f5fbdea0d8d5b0247eea3eae0064d604fe37fcbd26ecfc861cc26d64b
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / python_cq-0.23.0-py3-none-any.whl
| Download URL | python_cq-0.23.0-py3-none-any.whl |
|---|---|
| Size | 22.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
9b1fed236f0fb9613966abcd717f34877774ce2cb23fb071bd2b8eda0ea6a668
|
|
BLAKE2b-256 checksum How to use checksums |
f48e906a30417f181023dc1b0d4271e5bb7b600029fe74f6fe8621330c98cffd
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.5 {"installer":{"name":"uv","version":"0.12.5","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
