Skip to content

BaseExceptionMiddleware

faststream.broker.middlewares.exception.BaseExceptionMiddleware #

BaseExceptionMiddleware(handlers, publish_handlers, msg=None)

Bases: BaseMiddleware

Source code in faststream/broker/middlewares/exception.py 
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    handlers: CastedHandlers,
    publish_handlers: CastedPublishingHandlers,
    msg: Optional[Any] = None,
) -> None:
    super().__init__(msg)
    self._handlers = handlers
    self._publish_handlers = publish_handlers

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

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

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)

consume_scope async #

consume_scope(call_next, msg)
Source code in faststream/broker/middlewares/exception.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def consume_scope(
    self,
    call_next: "AsyncFuncAny",
    msg: "StreamMessage[Any]",
) -> Any:
    try:
        return await call_next(await self.on_consume(msg))

    except Exception as exc:
        exc_type = type(exc)

        for handler_type, handler in self._publish_handlers:
            if issubclass(exc_type, handler_type):
                return await handler(exc)

        raise exc

after_processed async #

after_processed(exc_type=None, exc_val=None, exc_tb=None)
Source code in faststream/broker/middlewares/exception.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def after_processed(
    self,
    exc_type: Optional[Type[BaseException]] = None,
    exc_val: Optional[BaseException] = None,
    exc_tb: Optional["TracebackType"] = None,
) -> Optional[bool]:
    if exc_type:
        for handler_type, handler in self._handlers:
            if issubclass(exc_type, handler_type):
                # TODO: remove it after context will be moved to middleware
                # In case parser/decoder error occurred
                scope: ContextManager[Any]
                if not context.get_local("message"):
                    scope = context.scope("message", self.msg)
                else:
                    scope = sync_fake_context()

                with scope:
                    await handler(exc_val)

                return True

        return False

    return None