Gate

Gate access to a middleware.

Usage

Gates access to a middleware behind a callback, if the result of the callback is True then the inner middleware will be inserted next into the chain, otherwise the chain will continue as normal.

The gate callback will be called with the current message.

Registration

Add the middleware to your bus.

def gate(message: banshee.Message[typing.Any]) -> bool:
   return isinstance(message.request, FooMixin)

bus = (
    banshee.builder()
    .with_middleware(banshee.GateMiddleware(FooHandler(), gate))
    .with_locator(registry)
    .build()
)

Reference

class banshee.Gate(*args, **kwargs)

Bases: Protocol

Gate.

Check whether the decorated middleware should be used.

__call__(message)

Check.

Check whether a message should be processed by the inner middleware. :param message: the message being processed :returns: whether to process the message

Parameters

message (Message[Any]) –

Return type

bool

class banshee.GateMiddleware(inner, gate)

Bases: Middleware

Gate middleware.

A Middleware decorator to conditionally invoke the decorated middleware depending on the result of a gate callback.

When the callback passes, decorated middleware will be inserted next in the chain before processing continues.

Parameters
Return type

None

async __call__(message, handle)

Handle message.

Forward the message to the inner middleware or the next handler in the chain based on the check callback result.

Parameters
Returns

processed message

Return type

Message[T]