On this page

• Usage
  • Registration
• Reference
  • Gate
  • GateMiddleware

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:

• inner (Middleware) – inner handler

• gate (Gate) – gate callback

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:

• message (Message[T]) – message to process

• handle (HandleMessage) – next middleware invoker

Returns:

processed message

Return type:

Message[T]