mediatpy package

Module contents

class mediatpy.Mediator(request_handler_factory=None, pipeline_behavior_factory=None, notification_handler_factory=None, raise_error_if_not_any_registered_notification_handler=False)

Bases: object

Class to register handlers and coordinate their execution in response to arrivals of requests and notifications

All parameters are optional.

my_mediator = Mediator()

Parameters:

request_handler_factory (Optional[Callable[[Type[RequestHandler]], Awaitable[RequestHandler]]]) – Custom RequestHandler factory
pipeline_behavior_factory (Optional[Callable[[Type[PipelineBehavior]], Awaitable[PipelineBehavior]]]) – Custom PipelineBehavior factory
notification_handler_factory (Optional[Callable[[Type[NotificationHandler]], Awaitable[NotificationHandler]]]) – Custom NotificationHandler factory
raise_error_if_not_any_registered_notification_handler (bool) – Raise an error if no NotificationHandler is found when a Notification is published

notification_handler(notification_handler)

Decorator to register a NotificationHandler

Return type:: None

pipeline_behavior(pipeline_behavior)

Decorator to register a PipelineBehavior

Return type:: None

async publish(notification)

Publish a Notification

my_notification = MyNotification()
await my_mediator.publish(my_notification)

Return type:: None

register_notification_handler(notification_handler)

Manual registration of a NotificationHandler

Return type:: None

register_pipeline_behavior(pipeline_behavior)

Manual registration of a PipelineBehavior

Return type:: None

register_request_handler(request_handler)

Manual registration of a RequestHandler

my_mediator.register_request_handler(MyRequestHandler)

Return type:: None

request_handler(request_handler)

Decorator to register a RequestHandler

@my_mediator.request_handler
class MyRequestHandler(RequestHandler[MyRequest, MyResponse]):
    async def handle(self, request: MyRequest) -> MyResponse:
        return MyResponse()

Return type:: None

async send(request)

Send a Request

my_request = MyRequest()
await my_mediator.send(my_request)

Return type:: TypeVar(TResponse)

exception mediatpy.NoRequestHandlerFoundError(request)

Bases: Exception

Error to indicate that a Request has not a registered RequestHandler to handle it

exception mediatpy.NotAnyNotificationHandlerFoundError(notification)

Bases: Exception

Error to indicate that a Notification has not any registered NotificationHandler to handle it

class mediatpy.Notification

Bases: object

Base class for any notification

class mediatpy.NotificationHandler

Bases: ABC, Generic[TNotification]

Base class for any notification handler

abstractmethod async handle(notification)

Return type:: None

class mediatpy.PipelineBehavior

Bases: ABC, Generic[TRequest, TResponse]

Base class for any pipeline behavior

abstractmethod async handle(request, next_behavior)

Abstract method to handle a request and to return a response

Parameters:

request (TypeVar(TRequest, bound= Request)) – Request to be handled
next_behavior (Callable[..., Awaitable[TypeVar(TResponse)]]) – Next pipeline behavior or request handler to execute

Return type:

TypeVar(TResponse)

class mediatpy.Request

Bases: Generic[TResponse]

Base class for any request

class mediatpy.RequestHandler

Bases: ABC, Generic[TRequest, TResponse]

Base class for any request handler

abstractmethod async handle(request)

Return type:: TypeVar(TResponse)