Metadata-Version: 2.1
Name: pycqs
Version: 1.0.1
Summary: Python Command Query Segregation (CQS) utilities
Author-Email: Raphael Castro <rafaphcastro@gmail.com>
License: MIT
Classifier: Development Status :: 5 - Production/Stable
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Information Technology
Classifier: Intended Audience :: System Administrators
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: Unix
Classifier: Operating System :: POSIX :: Linux
Classifier: Environment :: Console
Classifier: Environment :: MacOS X
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet
Project-URL: Homepage, https://github.com/rafaph/pycqs
Project-URL: Documentation, https://github.com/rafaph/pycqs
Project-URL: Source, https://github.com/rafaph/pycqs
Project-URL: Changelog, https://github.com/rafaph/pycqs/blob/main/HISTORY.md
Requires-Python: >=3.10
Description-Content-Type: text/markdown

# 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.:

```sh
pip install pycqs
```

## Usage

- Using the `CommandBus`.

```python
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](https://github.com/rafaph/pycqs/blob/main/tests/test_command_bus.py).

- Using the `QueryBus`.

```python
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](https://github.com/rafaph/pycqs/blob/main/tests/test_query_bus.py).

## License

MIT
