Skip to content

ExceptionMiddleware

faststream.middlewares.ExceptionMiddleware #

ExceptionMiddleware(handlers=None, publish_handlers=None)
Source code in faststream/middlewares/exception.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    handlers: dict[type[Exception], GeneralExceptionHandler] | None = None,
    publish_handlers: dict[type[Exception], PublishingExceptionHandler] | None = None,
) -> None:
    self._handlers: CastedHandlers = [
        (IgnoredException, ignore_handler),
        *(
            (
                exc_type,
                apply_types(
                    cast("Callable[..., Awaitable[None]]", to_async(handler)),
                    serializer_cls=None,
                ),
            )
            for exc_type, handler in (handlers or {}).items()
        ),
    ]

    self._publish_handlers: CastedPublishingHandlers = [
        (IgnoredException, ignore_handler),
        *(
            (exc_type, apply_types(to_async(handler), serializer_cls=None))
            for exc_type, handler in (publish_handlers or {}).items()
        ),
    ]

add_handler #

add_handler(
    exc: type[Exception], publish: Literal[False] = False
) -> Callable[
    [GeneralExceptionHandler], GeneralExceptionHandler
]

add_handler(
    exc: type[Exception], publish: Literal[True] = ...
) -> Callable[
    [PublishingExceptionHandler], PublishingExceptionHandler
]

add_handler(exc, publish=False)
Source code in faststream/middlewares/exception.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def add_handler(
    self,
    exc: type[Exception],
    publish: bool = False,
) -> (
    Callable[[GeneralExceptionHandler], GeneralExceptionHandler]
    | Callable[[PublishingExceptionHandler], PublishingExceptionHandler]
):
    if publish:

        def pub_wrapper(
            func: PublishingExceptionHandler,
        ) -> PublishingExceptionHandler:
            self._publish_handlers.append(
                (
                    exc,
                    apply_types(to_async(func), serializer_cls=None),
                ),
            )
            return func

        return pub_wrapper

    def default_wrapper(
        func: GeneralExceptionHandler,
    ) -> GeneralExceptionHandler:
        self._handlers.append(
            (
                exc,
                apply_types(to_async(func), serializer_cls=None),
            ),
        )
        return func

    return default_wrapper