Skip to content

BaseMiddleware

faststream.BaseMiddleware #

BaseMiddleware(
    msg: AnyMsg | None, /, *, context: ContextRepo
)

Bases: Generic[PublishCommandType, AnyMsg]

A base middleware class.

Source code in faststream/_internal/middlewares.py 
19
20
21
22
23
24
25
26
27
def __init__(
    self,
    msg: AnyMsg | None,
    /,
    *,
    context: "ContextRepo",
) -> None:
    self.msg = msg
    self.context = context

msg instance-attribute #

msg = msg

context instance-attribute #

context = context

on_receive async #

on_receive() -> None

Hook to call on message receive.

Source code in faststream/_internal/middlewares.py 
29
30
async def on_receive(self) -> None:
    """Hook to call on message receive."""

after_processed async #

after_processed(
    exc_type: type[BaseException] | None = None,
    exc_val: BaseException | None = None,
    exc_tb: Optional[TracebackType] = None,
) -> bool | None

Asynchronously called after processing.

Source code in faststream/_internal/middlewares.py 
32
33
34
35
36
37
38
39
async def after_processed(
    self,
    exc_type: type[BaseException] | None = None,
    exc_val: BaseException | None = None,
    exc_tb: Optional["TracebackType"] = None,
) -> bool | None:
    """Asynchronously called after processing."""
    return False

consume_scope async #

consume_scope(
    call_next: AsyncFuncAny, msg: StreamMessage[AnyMsg]
) -> Any

Asynchronously consumes a message and returns an asynchronous iterator of decoded messages.

Source code in faststream/_internal/middlewares.py 
54
55
56
57
58
59
60
async def consume_scope(
    self,
    call_next: "AsyncFuncAny",
    msg: "StreamMessage[AnyMsg]",
) -> Any:
    """Asynchronously consumes a message and returns an asynchronous iterator of decoded messages."""
    return await call_next(msg)

publish_scope async #

publish_scope(
    call_next: Callable[
        [PublishCommandType], Awaitable[Any]
    ],
    cmd: PublishCommandType,
) -> Any

Publish a message and return an async iterator.

Source code in faststream/_internal/middlewares.py 
62
63
64
65
66
67
68
async def publish_scope(
    self,
    call_next: Callable[[PublishCommandType], Awaitable[Any]],
    cmd: PublishCommandType,
) -> Any:
    """Publish a message and return an async iterator."""
    return await call_next(cmd)