pycqs

Command Query Segregation (CQS) utilities for Python.

Provides the CommandBus and QueryBus classes.

Requirements

Python 3.10 or above.

Installation

You can use pip to install pycqs with e.g.:

pip install pycqs

Usage

• Using the CommandBus.

import asyncio
from dataclasses import dataclass

from pycqs import (
    Command,
    CommandBus,
    CommandHandler,
)


# Create the command
@dataclass
class HelloWorldCommand(Command):
    name: str


# Create the command handler
class HelloWorldCommandHandler(
    CommandHandler[HelloWorldCommand],
):
    async def execute(
        self,
        command: HelloWorldCommand,
    ) -> None:
        print(f'Hello World, {command.name}')


async def main() -> None:
    # Initialize the command bus
    command_bus = CommandBus()
    command_bus.register_handler(
        HelloWorldCommand,
        HelloWorldCommandHandler(),
    )

    # Execute a command
    command = HelloWorldCommand('cqs')
    await command_bus.execute(command)


if __name__ == '__main__':
    asyncio.run(main())

More usage examples on tests.

• Using the QueryBus.

import asyncio
from dataclasses import dataclass

from pycqs import (
    Query,
    QueryBus,
    QueryHandler,
    QueryResult,
)


# Create the query
@dataclass
class HelloWorldQuery(Query):
    name: str


# Create the query result
@dataclass
class HelloWorldQueryResult(QueryResult):
    message: str


# Create the query handler
class HelloWorldQueryHandler(
    QueryHandler[
        HelloWorldQuery,
        HelloWorldQueryResult,
    ],
):
    async def execute(
        self,
        query: HelloWorldQuery,
    ) -> HelloWorldQueryResult:
        message = f'Hello World, {query.name}'
        return HelloWorldQueryResult(message)


async def main() -> None:
    # Initialize the query bus
    query_bus = QueryBus()
    query_bus.register_handler(
        HelloWorldQuery,
        HelloWorldQueryHandler(),
    )

    # Execute a query
    query = HelloWorldQuery('cqs')
    query_result: HelloWorldQueryResult = await query_bus.execute(query)
    print(query_result.message)


if __name__ == '__main__':
    asyncio.run(main())

More usage examples on tests.

License

MIT