gridflow_python_mediator.mediator

Core mediator for request/response dispatch.

Extracted from GridFlow backend/src/apps/token_generator/application/common/mediator/mediator.py

Attributes

TRequest
TResult

Classes

Mediator

Generic request/response dispatcher (Mediator pattern).

Module Contents

gridflow_python_mediator.mediator.TRequest

gridflow_python_mediator.mediator.TResult

class gridflow_python_mediator.mediator.Mediator

Generic request/response dispatcher (Mediator pattern).

Decouples senders from handlers, allowing for: - Pipeline behaviors (logging, timing, validation) - Dynamic handler registration - Type-safe dispatch

Example:

>>> mediator = Mediator()
>>> mediator.register_handler(MyRequest, MyHandler())
>>> result = await mediator.send(MyRequest(...))

register_handler(request_type: Type[TRequest], handler: Any) → None

Register handler for request type.

Args:

request_type: Request/Command/Query type handler: Handler instance with handle() method

Raises:

ValueError: If handler already registered for type

Example:

>>> mediator.register_handler(CreateUserCommand, CreateUserHandler())

async send(request: TRequest) → TResult

Dispatch request to its registered handler.

Executes pipeline behaviors before calling handler.

Args:

request: Request/Command/Query to send

Returns:

Result from handler

Raises:

ValueError: If no handler registered for request type

Example:

>>> cmd = CreateUserCommand(name="John")
>>> user_id = await mediator.send(cmd)

add_pipeline_behavior(behavior: Callable) → None

Add pipeline behavior for cross-cutting concerns.

Behaviors are executed in order before handler. Can implement logging, timing, validation, etc.

Args:

behavior: Async function(request, handler) -> Optional[result]

Example:

>>> async def logging_behavior(request, handler):
...     print(f"Handling {type(request).__name__}")
...     return None  # Continue to handler
>>>
>>> mediator.add_pipeline_behavior(logging_behavior)