python_cqrs_dispatcher.dispatcher

CQRS dispatcher integrating CQRS core with mediator pattern.

Classes

CQRSDispatcher

Dispatcher specialized for CQRS pattern.

Module Contents

class python_cqrs_dispatcher.dispatcher.CQRSDispatcher

Dispatcher specialized for CQRS pattern.

Integrates python-cqrs-core with gridflow-python-mediator, providing: - Type-safe command/query dispatch - Separate command/query handler registration - Pipeline behavior support through mediator

Example:

>>> dispatcher = CQRSDispatcher()
>>> dispatcher.register_command_handler(
...     CreateUserCommand,
...     CreateUserHandler()
... )
>>> user_id = await dispatcher.send_command(cmd)

register_command_handler(command_type: Type[python_cqrs_core.ICommand], handler: python_cqrs_core.ICommandHandler) → None

Register type-safe command handler.

Args:

command_type: Command class handler: Command handler instance

Raises:

ValueError: If handler already registered

Example:

>>> dispatcher.register_command_handler(
...     CreateUserCommand,
...     CreateUserHandler()
... )

register_query_handler(query_type: Type[python_cqrs_core.IQuery], handler: python_cqrs_core.IQueryHandler) → None

Register type-safe query handler.

Args:

query_type: Query class handler: Query handler instance

Raises:

ValueError: If handler already registered

Example:

>>> dispatcher.register_query_handler(
...     GetUserQuery,
...     GetUserHandler()
... )

async send_command(command: python_cqrs_core.ICommand) → Any

Dispatch command with validation.

Args:

command: Command to dispatch

Returns:

Command result

Raises:

ValueError: If no handler registered

Example:

>>> cmd = CreateUserCommand(name="John")
>>> user_id = await dispatcher.send_command(cmd)

async send_query(query: python_cqrs_core.IQuery) → Any

Dispatch query with validation.

Args:

query: Query to dispatch

Returns:

Query result

Raises:

ValueError: If no handler registered

Example:

>>> query = GetUserQuery(user_id=1)
>>> user = await dispatcher.send_query(query)

add_pipeline_behavior(behavior) → None

Add pipeline behavior to mediator.

Behaviors execute before all command/query handlers.

Args:

behavior: Pipeline behavior function

Example:

>>> from gridflow_python_mediator import LoggingBehavior
>>> dispatcher.add_pipeline_behavior(
...     LoggingBehavior().handle
... )