ExceptionMiddleware

faststream.ExceptionMiddleware #

ExceptionMiddleware(handlers=None, publish_handlers=None)

Source code in faststream/broker/middlewares/exception.py

def __init__(
    self,
    handlers: Optional[
        Dict[
            Type[Exception],
            GeneralExceptionHandler,
        ]
    ] = None,
    publish_handlers: Optional[
        Dict[
            Type[Exception],
            PublishingExceptionHandler,
        ]
    ] = None,
) -> None:
    self._handlers: CastedHandlers = [
        (IgnoredException, ignore_handler),
        *(
            (
                exc_type,
                apply_types(
                    cast(Callable[..., Awaitable[None]], to_async(handler))
                ),
            )
            for exc_type, handler in (handlers or {}).items()
        ),
    ]

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

add_handler #

add_handler(exc, publish=False)

Source code in faststream/broker/middlewares/exception.py

def add_handler(
    self,
    exc: Type[Exception],
    publish: bool = False,
) -> Union[
    Callable[[GeneralExceptionHandler], GeneralExceptionHandler],
    Callable[[PublishingExceptionHandler], PublishingExceptionHandler],
]:
    if publish:

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

        return pub_wrapper

    else:

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

        return default_wrapper