Skip to content

ExceptionMiddleware

faststream.middlewares.exception.ExceptionMiddleware #

ExceptionMiddleware(
    handlers: dict[type[Exception], GeneralExceptionHandler]
    | None = None,
    publish_handlers: dict[
        type[Exception], PublishingExceptionHandler
    ]
    | None = None,
)
Source code in faststream/middlewares/exception.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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: type[Exception], publish: bool = False
) -> (
    Callable[
        [GeneralExceptionHandler], GeneralExceptionHandler
    ]
    | Callable[
        [PublishingExceptionHandler],
        PublishingExceptionHandler,
    ]
)
Source code in faststream/middlewares/exception.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
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[exc] = apply_types(
                to_async(func),
                serializer_cls=None,
            )
            return func

        return pub_wrapper

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

    return default_wrapper