Skip to content

BaseMiddleware

faststream.BaseMiddleware #

BaseMiddleware(msg=None)

A base middleware class.

Source code in faststream/broker/middlewares/base.py 
15
16
def __init__(self, msg: Optional[Any] = None) -> None:
    self.msg = msg

msg instance-attribute #

msg = msg

on_receive async #

on_receive()

Hook to call on message receive.

Source code in faststream/broker/middlewares/base.py 
18
19
20
async def on_receive(self) -> None:
    """Hook to call on message receive."""
    pass

after_processed async #

after_processed(exc_type=None, exc_val=None, exc_tb=None)

Asynchronously called after processing.

Source code in faststream/broker/middlewares/base.py 
22
23
24
25
26
27
28
29
async def after_processed(
    self,
    exc_type: Optional[Type[BaseException]] = None,
    exc_val: Optional[BaseException] = None,
    exc_tb: Optional["TracebackType"] = None,
) -> Optional[bool]:
    """Asynchronously called after processing."""
    return False

on_consume async #

on_consume(msg)

Asynchronously consumes a message.

Source code in faststream/broker/middlewares/base.py 
44
45
46
47
48
49
async def on_consume(
    self,
    msg: "StreamMessage[Any]",
) -> "StreamMessage[Any]":
    """Asynchronously consumes a message."""
    return msg

after_consume async #

after_consume(err)

A function to handle the result of consuming a resource asynchronously.

Source code in faststream/broker/middlewares/base.py 
51
52
53
54
async def after_consume(self, err: Optional[Exception]) -> None:
    """A function to handle the result of consuming a resource asynchronously."""
    if err is not None:
        raise err

consume_scope async #

consume_scope(call_next, msg)

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

Source code in faststream/broker/middlewares/base.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
async def consume_scope(
    self,
    call_next: "AsyncFuncAny",
    msg: "StreamMessage[Any]",
) -> Any:
    """Asynchronously consumes a message and returns an asynchronous iterator of decoded messages."""
    err: Optional[Exception] = None
    try:
        result = await call_next(await self.on_consume(msg))

    except Exception as e:
        err = e

    else:
        return result

    finally:
        await self.after_consume(err)

on_publish async #

on_publish(msg, *args, **kwargs)

Asynchronously handle a publish event.

Source code in faststream/broker/middlewares/base.py 
75
76
77
78
79
80
81
82
async def on_publish(
    self,
    msg: Any,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Asynchronously handle a publish event."""
    return msg

after_publish async #

after_publish(err)

Asynchronous function to handle the after publish event.

Source code in faststream/broker/middlewares/base.py 
84
85
86
87
88
89
90
async def after_publish(
    self,
    err: Optional[Exception],
) -> None:
    """Asynchronous function to handle the after publish event."""
    if err is not None:
        raise err

publish_scope async #

publish_scope(call_next, msg, *args, **kwargs)

Publish a message and return an async iterator.

Source code in faststream/broker/middlewares/base.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def publish_scope(
    self,
    call_next: "AsyncFunc",
    msg: Any,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """Publish a message and return an async iterator."""
    err: Optional[Exception] = None
    try:
        result = await call_next(
            await self.on_publish(msg, *args, **kwargs),
            *args,
            **kwargs,
        )

    except Exception as e:
        err = e

    else:
        return result

    finally:
        await self.after_publish(err)