From c7d7825b856b6816047366c86c2e3c8bb21d114d Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Sat, 26 Sep 2026 22:27:16 +0300 Subject: [PATCH 1/7] feat(kafka): add subscriber and broker exception handlers --- faststream/_internal/configs/broker.py | 7 ++- faststream/_internal/configs/endpoint.py | 3 +- .../_internal/endpoint/subscriber/usecase.py | 21 +++++++- faststream/_internal/types.py | 4 ++ faststream/kafka/broker/broker.py | 7 +++ faststream/kafka/broker/registrator.py | 11 ++++ faststream/kafka/subscriber/factory.py | 3 ++ tests/brokers/base/exception_handlers.py | 50 +++++++++++++++++++ .../brokers/kafka/test_exception_handlers.py | 10 ++++ 9 files changed, 112 insertions(+), 4 deletions(-) create mode 100644 tests/brokers/base/exception_handlers.py create mode 100644 tests/brokers/kafka/test_exception_handlers.py diff --git a/faststream/_internal/configs/broker.py b/faststream/_internal/configs/broker.py index 82355521b8d..61763d62341 100644 --- a/faststream/_internal/configs/broker.py +++ b/faststream/_internal/configs/broker.py @@ -16,7 +16,11 @@ from faststream._internal.context import ContextRepo from faststream._internal.parser import CodecProto - from faststream._internal.types import BrokerMiddleware, CustomCallable + from faststream._internal.types import ( + BrokerMiddleware, + CustomCallable, + ExceptionHandler, + ) from faststream.middlewares import AckPolicy @@ -29,6 +33,7 @@ class BrokerConfig: broker_parser: Optional["CustomCallable"] = None broker_decoder: Optional["CustomCallable"] = None broker_codec: Optional["CodecProto"] = None + broker_exception_handler: Optional["ExceptionHandler"] = None producer: "ProducerProto[Any]" = field(default_factory=ProducerUnset) logger: "LoggerState" = field(default_factory=LoggerState) diff --git a/faststream/_internal/configs/endpoint.py b/faststream/_internal/configs/endpoint.py index 6beaf99d548..7ac7cd4a983 100644 --- a/faststream/_internal/configs/endpoint.py +++ b/faststream/_internal/configs/endpoint.py @@ -5,7 +5,7 @@ from faststream.middlewares import AckPolicy if TYPE_CHECKING: - from faststream._internal.types import AsyncCallable + from faststream._internal.types import AsyncCallable, ExceptionHandler from .broker import BrokerConfig @@ -23,6 +23,7 @@ class PublisherUsecaseConfig(EndpointConfig): @dataclass(kw_only=True) class SubscriberUsecaseConfig(EndpointConfig): no_reply: bool = False + exception_handler: "ExceptionHandler | None" = None _ack_policy: AckPolicy = field(default_factory=lambda: EMPTY, repr=False) diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index 3bba03fcf9c..f700587f860 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -45,6 +45,7 @@ from faststream._internal.endpoint.publisher import PublisherProto from faststream._internal.parser import CodecProto from faststream._internal.types import ( + AsyncExceptionHandler, AsyncFilter, BrokerMiddleware, CustomCallable, @@ -87,6 +88,11 @@ def __init__( self._no_reply = config.no_reply self._parser = config.parser self._decoder = config.decoder + self._exception_handler: AsyncExceptionHandler | None = ( + to_async(config.exception_handler) + if config.exception_handler is not None + else None + ) self.ack_policy = config.ack_policy self.__auto_ack_disabled = config.auto_ack_disabled @@ -313,6 +319,17 @@ def real_wrapper( return real_wrapper(func) + async def _handle_exception(self, exc: BaseException) -> bool: + if self._exception_handler is not None and await self._exception_handler(exc): + return True + + broker_handler = self._outer_config.broker_exception_handler + if broker_handler is not None: + async_handler: AsyncExceptionHandler = to_async(broker_handler) + return await async_handler(exc) + + return False + async def consume(self, msg: MsgType) -> Any: """Consume a message asynchronously.""" if not self.running: @@ -332,9 +349,9 @@ async def consume(self, msg: MsgType) -> Any: if app := self._outer_config.context.get("app"): app.exit() - except Exception: # nosec B110 # noqa: S110 + except Exception as exc: # All other exceptions were logged by CriticalLogMiddleware - pass + await self._handle_exception(exc) async def process_message(self, msg: MsgType) -> "Response": """Execute all message processing stages.""" diff --git a/faststream/_internal/types.py b/faststream/_internal/types.py index e4835d1b8a5..c0c5580a7da 100644 --- a/faststream/_internal/types.py +++ b/faststream/_internal/types.py @@ -42,6 +42,10 @@ AsyncFilter: TypeAlias = Callable[[StreamMsg], Awaitable[bool]] Filter: TypeAlias = SyncFilter[StreamMsg] | AsyncFilter[StreamMsg] +SyncExceptionHandler: TypeAlias = Callable[[BaseException], bool] +AsyncExceptionHandler: TypeAlias = Callable[[BaseException], Awaitable[bool]] +ExceptionHandler: TypeAlias = SyncExceptionHandler | AsyncExceptionHandler + SyncCallable: TypeAlias = Callable[ [Any], Any, diff --git a/faststream/kafka/broker/broker.py b/faststream/kafka/broker/broker.py index 1f537bd0744..a8b3a65ae86 100644 --- a/faststream/kafka/broker/broker.py +++ b/faststream/kafka/broker/broker.py @@ -59,6 +59,7 @@ from faststream._internal.types import ( BrokerMiddleware, CustomCallable, + ExceptionHandler, ) from faststream.kafka.message import KafkaMessage from faststream.kafka.types import KafkaSendableMessage @@ -234,6 +235,7 @@ def __init__( decoder: Optional["CustomCallable"] = None, codec: Optional["CodecProto"] = None, parser: Optional["CustomCallable"] = None, + exception_handler: "ExceptionHandler | None" = None, dependencies: Sequence["Dependant"] = (), middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (), routers: Iterable[KafkaRegistrator] = (), @@ -348,6 +350,10 @@ def __init__( Custom codec object. parser (Optional[CustomCallable]): Custom parser object. + exception_handler (ExceptionHandler | None): + Sync or async exception handler called when a subscriber's + exception handler is absent or returns False. + Return True if handled, otherwise False for default handling. dependencies (Sequence[Dependant]): Dependencies to apply to all broker subscribers. middlewares (Sequence[BrokerMiddlewarep[Any, Any]]): @@ -455,6 +461,7 @@ def __init__( broker_decoder=decoder, broker_codec=codec, broker_parser=parser, + broker_exception_handler=exception_handler, broker_middlewares=middlewares, logger=make_kafka_logger_state( logger=logger, diff --git a/faststream/kafka/broker/registrator.py b/faststream/kafka/broker/registrator.py index c2bf01137d7..c98e7ed73dc 100644 --- a/faststream/kafka/broker/registrator.py +++ b/faststream/kafka/broker/registrator.py @@ -30,6 +30,7 @@ from faststream._internal.types import ( BrokerMiddleware, CustomCallable, + ExceptionHandler, ) from faststream.kafka.publisher.usecase import ( BatchPublisher, @@ -96,6 +97,7 @@ def subscriber( max_workers: None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -146,6 +148,7 @@ def subscriber( max_workers: None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -196,6 +199,7 @@ def subscriber( max_workers: int = ..., ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -246,6 +250,7 @@ def subscriber( max_workers: int = ..., ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -296,6 +301,7 @@ def subscriber( max_workers: int | None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -351,6 +357,7 @@ def subscriber( max_workers: int | None = None, ack_policy: AckPolicy = EMPTY, no_reply: bool = False, + exception_handler: "ExceptionHandler | None" = None, # Specification args title: str | None = None, description: str | None = None, @@ -552,6 +559,9 @@ def subscriber( no_ack: Whether to disable **FastStream** auto acknowledgement logic or not. ack_policy: Acknowledgement policy for the subscriber. no_reply: Whether to disable **FastStream** RPC and Reply To auto responses or not. + exception_handler: Sync or async exception handler called before + the broker's handler. Return True if handled, otherwise False + to delegate to the broker. title: Specification subscriber object title. description: Specification subscriber object description. " "Uses decorated docstring as default. include_in_schema: Whetever to include operation in Specification schema or not. @@ -595,6 +605,7 @@ def subscriber( partitions=partitions, ack_policy=ack_policy, no_reply=no_reply, + exception_handler=exception_handler, config=cast("KafkaBrokerConfig", self.config), # Specification title_=title, diff --git a/faststream/kafka/subscriber/factory.py b/faststream/kafka/subscriber/factory.py index fe64c506bf0..5bb2133566e 100644 --- a/faststream/kafka/subscriber/factory.py +++ b/faststream/kafka/subscriber/factory.py @@ -19,6 +19,7 @@ if TYPE_CHECKING: from aiokafka.abc import ConsumerRebalanceListener + from faststream._internal.types import ExceptionHandler from faststream.kafka.configs import KafkaBrokerConfig from faststream.kafka.schemas import TopicPartition @@ -38,6 +39,7 @@ def create_subscriber( ack_policy: "AckPolicy", max_workers: int, no_reply: bool, + exception_handler: "ExceptionHandler | None", config: "KafkaBrokerConfig", # Specification args title_: str | None, @@ -66,6 +68,7 @@ def create_subscriber( listener=listener, pattern=pattern, no_reply=no_reply, + exception_handler=exception_handler, _outer_config=config, _ack_policy=ack_policy, ) diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py new file mode 100644 index 00000000000..cf83993df5d --- /dev/null +++ b/tests/brokers/base/exception_handlers.py @@ -0,0 +1,50 @@ +from typing import Any + +import pytest + +from tests.brokers.base.basic import BaseTestcaseConfig + + +@pytest.mark.asyncio() +class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): + @pytest.mark.parametrize( + ("local_result", "broker_result", "expected_result", "expected_calls"), + ( + (True, False, True, ["subscriber"]), + (False, True, True, ["subscriber", "broker"]), + (False, False, False, ["subscriber", "broker"]), + ), + ) + async def test_exception_handler_chain( + self, + queue: str, + local_result: bool, + broker_result: bool, + expected_result: bool, + expected_calls: list[str], + ) -> None: + error = ValueError("test error") + calls: list[tuple[str, BaseException]] = [] + + def local_handler(exc: BaseException) -> bool: + calls.append(("subscriber", exc)) + return local_result + + async def broker_handler(exc: BaseException) -> bool: + calls.append(("broker", exc)) + return broker_result + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=local_handler, + ) + + result = await subscriber._handle_exception(error) + + assert (result, calls) == ( + expected_result, + [(name, error) for name in expected_calls], + ) diff --git a/tests/brokers/kafka/test_exception_handlers.py b/tests/brokers/kafka/test_exception_handlers.py new file mode 100644 index 00000000000..cef1e4e52c2 --- /dev/null +++ b/tests/brokers/kafka/test_exception_handlers.py @@ -0,0 +1,10 @@ +import pytest + +from tests.brokers.base.exception_handlers import ExceptionHandlersTestcase + +from .basic import KafkaTestcaseConfig + + +@pytest.mark.kafka() +class TestExceptionHandlers(KafkaTestcaseConfig, ExceptionHandlersTestcase): + pass From 9654b803f7e737696b50e4dca9bfd49fe897c209 Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Sun, 27 Sep 2026 01:35:57 +0300 Subject: [PATCH 2/7] test: cover exception handler dispatch from consume --- tests/brokers/base/exception_handlers.py | 32 ++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py index cf83993df5d..09e0b79197c 100644 --- a/tests/brokers/base/exception_handlers.py +++ b/tests/brokers/base/exception_handlers.py @@ -1,4 +1,5 @@ from typing import Any +from unittest.mock import AsyncMock import pytest @@ -48,3 +49,34 @@ async def broker_handler(exc: BaseException) -> bool: expected_result, [(name, error) for name in expected_calls], ) + + async def test_consume_calls_exception_handler( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + received_errors: list[BaseException] = [] + + async def exception_handler(exc: BaseException) -> bool: + received_errors.append(exc) + return True + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=exception_handler, + ) + + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert received_errors == [error] From 4ac61ad4d19315223c3ee916b6367eed4127aefe Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Mon, 28 Sep 2026 21:07:38 +0300 Subject: [PATCH 3/7] fix: support exception handler context and propagate unhandled errors --- faststream/_internal/configs/broker.py | 12 + faststream/_internal/configs/endpoint.py | 16 +- .../_internal/endpoint/subscriber/usecase.py | 49 ++- faststream/confluent/subscriber/config.py | 1 + faststream/kafka/subscriber/config.py | 1 + tests/brokers/base/exception_handlers.py | 413 ++++++++++++++---- 6 files changed, 399 insertions(+), 93 deletions(-) diff --git a/faststream/_internal/configs/broker.py b/faststream/_internal/configs/broker.py index 61763d62341..b668db8b200 100644 --- a/faststream/_internal/configs/broker.py +++ b/faststream/_internal/configs/broker.py @@ -9,6 +9,7 @@ from faststream._internal.logger import LoggerState from faststream._internal.producer import ProducerProto, ProducerUnset from faststream._internal.types import IdGenerator +from faststream._internal.utils.functions import to_async from faststream.message import gen_cor_id if TYPE_CHECKING: @@ -17,6 +18,7 @@ from faststream._internal.context import ContextRepo from faststream._internal.parser import CodecProto from faststream._internal.types import ( + AsyncExceptionHandler, BrokerMiddleware, CustomCallable, ExceptionHandler, @@ -34,6 +36,11 @@ class BrokerConfig: broker_decoder: Optional["CustomCallable"] = None broker_codec: Optional["CodecProto"] = None broker_exception_handler: Optional["ExceptionHandler"] = None + _broker_exception_handler: Optional["AsyncExceptionHandler"] = field( + default=None, + init=False, + repr=False, + ) producer: "ProducerProto[Any]" = field(default_factory=ProducerUnset) logger: "LoggerState" = field(default_factory=LoggerState) @@ -50,6 +57,11 @@ def __post_init__(self) -> None: # untyped callers still pass a generator: the first subscriber would spend it self.broker_dependencies = tuple(self.broker_dependencies) + if self.broker_exception_handler is not None: + self._broker_exception_handler = to_async( + self.broker_exception_handler, + ) + def __repr__(self) -> str: return f"{self.__class__.__name__}(id: {id(self)})" diff --git a/faststream/_internal/configs/endpoint.py b/faststream/_internal/configs/endpoint.py index 7ac7cd4a983..3a962b51959 100644 --- a/faststream/_internal/configs/endpoint.py +++ b/faststream/_internal/configs/endpoint.py @@ -2,10 +2,15 @@ from typing import TYPE_CHECKING from faststream._internal.constants import EMPTY +from faststream._internal.utils.functions import to_async from faststream.middlewares import AckPolicy if TYPE_CHECKING: - from faststream._internal.types import AsyncCallable, ExceptionHandler + from faststream._internal.types import ( + AsyncCallable, + AsyncExceptionHandler, + ExceptionHandler, + ) from .broker import BrokerConfig @@ -24,12 +29,21 @@ class PublisherUsecaseConfig(EndpointConfig): class SubscriberUsecaseConfig(EndpointConfig): no_reply: bool = False exception_handler: "ExceptionHandler | None" = None + _exception_handler: "AsyncExceptionHandler | None" = field( + default=None, + init=False, + repr=False, + ) _ack_policy: AckPolicy = field(default_factory=lambda: EMPTY, repr=False) parser: "AsyncCallable" = field(init=False) decoder: "AsyncCallable" = field(init=False) + def __post_init__(self) -> None: + if self.exception_handler is not None: + self._exception_handler = to_async(self.exception_handler) + @property def auto_ack_disabled(self) -> bool: return self.ack_policy is AckPolicy.MANUAL diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index f700587f860..458b9e02a89 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -24,6 +24,7 @@ P_HandlerParams, T_HandlerReturn, ) +from faststream._internal.utils import apply_types from faststream._internal.utils.functions import FakeContext, to_async from faststream.exceptions import StopConsume, SubscriberNotFound from faststream.middlewares import AcknowledgementMiddleware @@ -88,11 +89,9 @@ def __init__( self._no_reply = config.no_reply self._parser = config.parser self._decoder = config.decoder - self._exception_handler: AsyncExceptionHandler | None = ( - to_async(config.exception_handler) - if config.exception_handler is not None - else None - ) + self._exception_handler: AsyncExceptionHandler | None = config._exception_handler + self._exception_handler_call: AsyncExceptionHandler | None = None + self._broker_exception_handler_call: AsyncExceptionHandler | None = None self.ack_policy = config.ack_policy self.__auto_ack_disabled = config.auto_ack_disabled @@ -202,6 +201,31 @@ def _get_parser_and_decoder( return async_parser, async_decoder def _build_fastdepends_model(self) -> None: + config = self._outer_config.fd_config + + self._exception_handler_call = ( + apply_types( + self._exception_handler, + serializer_cls=config._serializer, + context__=config.context, + ) + if self._exception_handler is not None + else None + ) + + broker_handler: AsyncExceptionHandler | None = ( + self._outer_config._broker_exception_handler + ) + self._broker_exception_handler_call = ( + apply_types( + broker_handler, + serializer_cls=config._serializer, + context__=config.context, + ) + if broker_handler is not None + else None + ) + for call in self.calls: async_parser, async_decoder = self._get_parser_and_decoder( call.item_parser, call.item_decoder @@ -320,13 +344,13 @@ def real_wrapper( return real_wrapper(func) async def _handle_exception(self, exc: BaseException) -> bool: - if self._exception_handler is not None and await self._exception_handler(exc): + local_handler = self._exception_handler_call + if local_handler is not None and await local_handler(exc): return True - broker_handler = self._outer_config.broker_exception_handler + broker_handler = self._broker_exception_handler_call if broker_handler is not None: - async_handler: AsyncExceptionHandler = to_async(broker_handler) - return await async_handler(exc) + return await broker_handler(exc) return False @@ -351,7 +375,12 @@ async def consume(self, msg: MsgType) -> Any: except Exception as exc: # All other exceptions were logged by CriticalLogMiddleware - await self._handle_exception(exc) + has_handler = ( + self._exception_handler_call is not None + or self._broker_exception_handler_call is not None + ) + if has_handler and not await self._handle_exception(exc): + raise async def process_message(self, msg: MsgType) -> "Response": """Execute all message processing stages.""" diff --git a/faststream/confluent/subscriber/config.py b/faststream/confluent/subscriber/config.py index 729c65c90b1..ae3f9f8ca11 100644 --- a/faststream/confluent/subscriber/config.py +++ b/faststream/confluent/subscriber/config.py @@ -31,6 +31,7 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): connection_data: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: + super().__post_init__() self.connection_data["enable_auto_commit"] = self.ack_first @property diff --git a/faststream/kafka/subscriber/config.py b/faststream/kafka/subscriber/config.py index 8c5eee0f73a..f276485237e 100644 --- a/faststream/kafka/subscriber/config.py +++ b/faststream/kafka/subscriber/config.py @@ -35,6 +35,7 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): partitions: Iterable["TopicPartition"] = field(default_factory=list) def __post_init__(self) -> None: + super().__post_init__() self.connection_args["enable_auto_commit"] = self.ack_first @property diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py index 09e0b79197c..56c892480d3 100644 --- a/tests/brokers/base/exception_handlers.py +++ b/tests/brokers/base/exception_handlers.py @@ -1,82 +1,331 @@ -from typing import Any -from unittest.mock import AsyncMock - -import pytest - -from tests.brokers.base.basic import BaseTestcaseConfig - - -@pytest.mark.asyncio() -class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): - @pytest.mark.parametrize( - ("local_result", "broker_result", "expected_result", "expected_calls"), - ( - (True, False, True, ["subscriber"]), - (False, True, True, ["subscriber", "broker"]), - (False, False, False, ["subscriber", "broker"]), - ), - ) - async def test_exception_handler_chain( - self, - queue: str, - local_result: bool, - broker_result: bool, - expected_result: bool, - expected_calls: list[str], - ) -> None: - error = ValueError("test error") - calls: list[tuple[str, BaseException]] = [] - - def local_handler(exc: BaseException) -> bool: - calls.append(("subscriber", exc)) - return local_result - - async def broker_handler(exc: BaseException) -> bool: - calls.append(("broker", exc)) - return broker_result - - broker = self.get_broker(exception_handler=broker_handler) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, - exception_handler=local_handler, - ) - - result = await subscriber._handle_exception(error) - - assert (result, calls) == ( - expected_result, - [(name, error) for name in expected_calls], - ) - - async def test_consume_calls_exception_handler( - self, - queue: str, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - error = ValueError("message processing failed") - received_errors: list[BaseException] = [] - - async def exception_handler(exc: BaseException) -> bool: - received_errors.append(exc) - return True - - broker = self.get_broker() - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, - exception_handler=exception_handler, - ) - - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) - - await subscriber.consume(object()) - - assert received_errors == [error] +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from faststream import Context +from tests.brokers.base.basic import BaseTestcaseConfig + + +@pytest.mark.asyncio() +class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): + async def test_subscriber_exception_handler_true_suppresses_error( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + received_errors: list[BaseException] = [] + + async def exception_handler(exc: BaseException) -> bool: + received_errors.append(exc) + return True + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=exception_handler, + ) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert received_errors == [error] + + async def test_subscriber_exception_handler_false_propagates_error( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + received_errors: list[BaseException] = [] + + async def exception_handler(exc: BaseException) -> bool: + received_errors.append(exc) + return False + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=exception_handler, + ) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + with pytest.raises( + ValueError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(object()) + + assert exc_info.value is error + assert received_errors == [error] + + async def test_subscriber_exception_handler_receives_context( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + expected_service = object() + received: list[tuple[BaseException, object]] = [] + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + received.append((exc, service)) + return True + + broker = self.get_broker(apply_types=True) + broker.context.set_global("error_service", expected_service) + + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=exception_handler, + ) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert received == [(error, expected_service)] + + async def test_subscriber_handles_error_without_calling_broker( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + calls: list[tuple[str, BaseException]] = [] + + async def local_handler(exc: BaseException) -> bool: + calls.append(("subscriber", exc)) + return True + + async def broker_handler(exc: BaseException) -> bool: + calls.append(("broker", exc)) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=local_handler, + ) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert calls == [("subscriber", error)] + + async def test_broker_handles_error_after_subscriber_declines( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + calls: list[str] = [] + + def local_handler(exc: BaseException) -> bool: + calls.append("subscriber") + return False + + def broker_handler(exc: BaseException) -> bool: + calls.append("broker") + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=local_handler, + ) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert calls == ["subscriber", "broker"] + + async def test_error_propagates_when_both_handlers_decline( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + calls: list[tuple[str, BaseException]] = [] + + async def local_handler(exc: BaseException) -> bool: + calls.append(("subscriber", exc)) + return False + + async def broker_handler(exc: BaseException) -> bool: + calls.append(("broker", exc)) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber( + *args, + **kwargs, + exception_handler=local_handler, + ) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + with pytest.raises( + ValueError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(object()) + + assert exc_info.value is error + assert calls == [ + ("subscriber", error), + ("broker", error), + ] + + async def test_broker_exception_handler_false_propagates_error( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + received_errors: list[BaseException] = [] + + async def exception_handler(exc: BaseException) -> bool: + received_errors.append(exc) + return False + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + with pytest.raises( + ValueError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(object()) + + assert exc_info.value is error + assert received_errors == [error] + + async def test_broker_exception_handler_true_suppresses_error( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + received_errors: list[BaseException] = [] + + async def exception_handler(exc: BaseException) -> bool: + received_errors.append(exc) + return True + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert received_errors == [error] + + async def test_broker_exception_handler_receives_context( + self, + queue: str, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + error = ValueError("message processing failed") + expected_service = object() + received: list[tuple[BaseException, object]] = [] + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + received.append((exc, service)) + return True + + broker = self.get_broker( + apply_types=True, + exception_handler=exception_handler, + ) + broker.context.set_global("error_service", expected_service) + + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + subscriber._build_fastdepends_model() + monkeypatch.setattr(subscriber, "running", True) + monkeypatch.setattr( + subscriber, + "process_message", + AsyncMock(side_effect=error), + ) + + await subscriber.consume(object()) + + assert received == [(error, expected_service)] From 878a2ffff4cd79a00d549849c4c6efe30ad4cfc6 Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Sat, 3 Oct 2026 15:11:45 +0300 Subject: [PATCH 4/7] refactor: prepare exception handlers in constructors and use Kafka e2e tests --- faststream/_internal/broker/broker.py | 11 + faststream/_internal/configs/broker.py | 6 - faststream/_internal/configs/endpoint.py | 16 +- .../_internal/endpoint/subscriber/usecase.py | 42 +-- faststream/confluent/subscriber/config.py | 1 - faststream/kafka/subscriber/config.py | 1 - tests/brokers/base/exception_handlers.py | 352 +++++++----------- .../brokers/kafka/test_exception_handlers.py | 149 +++++++- 8 files changed, 305 insertions(+), 273 deletions(-) diff --git a/faststream/_internal/broker/broker.py b/faststream/_internal/broker/broker.py index a4f49acf057..469d569818b 100644 --- a/faststream/_internal/broker/broker.py +++ b/faststream/_internal/broker/broker.py @@ -11,6 +11,7 @@ ConnectionType, MsgType, ) +from faststream._internal.utils import apply_types, to_async from .pub_base import BrokerPublishMixin from .registrator import Registrator @@ -21,6 +22,7 @@ from faststream._internal.context.repository import ContextRepo from faststream._internal.di import FastDependsConfig from faststream._internal.producer import ProducerProto + from faststream._internal.types import AsyncExceptionHandler from faststream.specification.schema import BrokerSpec @@ -48,6 +50,15 @@ def __init__( routers=routers, config=config, ) + if config.broker_exception_handler is not None: + async_handler: AsyncExceptionHandler = to_async( + config.broker_exception_handler, + ) + config._broker_exception_handler = apply_types( + async_handler, + serializer_cls=config.fd_config._serializer, + context__=config.context, + ) self.specification = specification self.running = False diff --git a/faststream/_internal/configs/broker.py b/faststream/_internal/configs/broker.py index b668db8b200..25c6b37770f 100644 --- a/faststream/_internal/configs/broker.py +++ b/faststream/_internal/configs/broker.py @@ -9,7 +9,6 @@ from faststream._internal.logger import LoggerState from faststream._internal.producer import ProducerProto, ProducerUnset from faststream._internal.types import IdGenerator -from faststream._internal.utils.functions import to_async from faststream.message import gen_cor_id if TYPE_CHECKING: @@ -57,11 +56,6 @@ def __post_init__(self) -> None: # untyped callers still pass a generator: the first subscriber would spend it self.broker_dependencies = tuple(self.broker_dependencies) - if self.broker_exception_handler is not None: - self._broker_exception_handler = to_async( - self.broker_exception_handler, - ) - def __repr__(self) -> str: return f"{self.__class__.__name__}(id: {id(self)})" diff --git a/faststream/_internal/configs/endpoint.py b/faststream/_internal/configs/endpoint.py index 3a962b51959..7ac7cd4a983 100644 --- a/faststream/_internal/configs/endpoint.py +++ b/faststream/_internal/configs/endpoint.py @@ -2,15 +2,10 @@ from typing import TYPE_CHECKING from faststream._internal.constants import EMPTY -from faststream._internal.utils.functions import to_async from faststream.middlewares import AckPolicy if TYPE_CHECKING: - from faststream._internal.types import ( - AsyncCallable, - AsyncExceptionHandler, - ExceptionHandler, - ) + from faststream._internal.types import AsyncCallable, ExceptionHandler from .broker import BrokerConfig @@ -29,21 +24,12 @@ class PublisherUsecaseConfig(EndpointConfig): class SubscriberUsecaseConfig(EndpointConfig): no_reply: bool = False exception_handler: "ExceptionHandler | None" = None - _exception_handler: "AsyncExceptionHandler | None" = field( - default=None, - init=False, - repr=False, - ) _ack_policy: AckPolicy = field(default_factory=lambda: EMPTY, repr=False) parser: "AsyncCallable" = field(init=False) decoder: "AsyncCallable" = field(init=False) - def __post_init__(self) -> None: - if self.exception_handler is not None: - self._exception_handler = to_async(self.exception_handler) - @property def auto_ack_disabled(self) -> bool: return self.ack_policy is AckPolicy.MANUAL diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index 458b9e02a89..dcfb5edeff0 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -89,9 +89,16 @@ def __init__( self._no_reply = config.no_reply self._parser = config.parser self._decoder = config.decoder - self._exception_handler: AsyncExceptionHandler | None = config._exception_handler self._exception_handler_call: AsyncExceptionHandler | None = None - self._broker_exception_handler_call: AsyncExceptionHandler | None = None + if config.exception_handler is not None: + async_handler: AsyncExceptionHandler = to_async( + config.exception_handler, + ) + self._exception_handler_call = apply_types( + async_handler, + serializer_cls=self._outer_config.fd_config._serializer, + context__=self._outer_config.context, + ) self.ack_policy = config.ack_policy self.__auto_ack_disabled = config.auto_ack_disabled @@ -201,31 +208,6 @@ def _get_parser_and_decoder( return async_parser, async_decoder def _build_fastdepends_model(self) -> None: - config = self._outer_config.fd_config - - self._exception_handler_call = ( - apply_types( - self._exception_handler, - serializer_cls=config._serializer, - context__=config.context, - ) - if self._exception_handler is not None - else None - ) - - broker_handler: AsyncExceptionHandler | None = ( - self._outer_config._broker_exception_handler - ) - self._broker_exception_handler_call = ( - apply_types( - broker_handler, - serializer_cls=config._serializer, - context__=config.context, - ) - if broker_handler is not None - else None - ) - for call in self.calls: async_parser, async_decoder = self._get_parser_and_decoder( call.item_parser, call.item_decoder @@ -348,7 +330,9 @@ async def _handle_exception(self, exc: BaseException) -> bool: if local_handler is not None and await local_handler(exc): return True - broker_handler = self._broker_exception_handler_call + broker_handler: AsyncExceptionHandler | None = ( + self._outer_config._broker_exception_handler + ) if broker_handler is not None: return await broker_handler(exc) @@ -377,7 +361,7 @@ async def consume(self, msg: MsgType) -> Any: # All other exceptions were logged by CriticalLogMiddleware has_handler = ( self._exception_handler_call is not None - or self._broker_exception_handler_call is not None + or self._outer_config._broker_exception_handler is not None ) if has_handler and not await self._handle_exception(exc): raise diff --git a/faststream/confluent/subscriber/config.py b/faststream/confluent/subscriber/config.py index ae3f9f8ca11..729c65c90b1 100644 --- a/faststream/confluent/subscriber/config.py +++ b/faststream/confluent/subscriber/config.py @@ -31,7 +31,6 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): connection_data: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: - super().__post_init__() self.connection_data["enable_auto_commit"] = self.ack_first @property diff --git a/faststream/kafka/subscriber/config.py b/faststream/kafka/subscriber/config.py index f276485237e..8c5eee0f73a 100644 --- a/faststream/kafka/subscriber/config.py +++ b/faststream/kafka/subscriber/config.py @@ -35,7 +35,6 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): partitions: Iterable["TopicPartition"] = field(default_factory=list) def __post_init__(self) -> None: - super().__post_init__() self.connection_args["enable_auto_commit"] = self.ack_first @property diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py index 56c892480d3..bca4abaf4b3 100644 --- a/tests/brokers/base/exception_handlers.py +++ b/tests/brokers/base/exception_handlers.py @@ -1,5 +1,6 @@ +import asyncio from typing import Any -from unittest.mock import AsyncMock +from unittest.mock import MagicMock, call import pytest @@ -9,304 +10,229 @@ @pytest.mark.asyncio() class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): + @pytest.mark.connected() async def test_subscriber_exception_handler_true_suppresses_error( self, queue: str, - monkeypatch: pytest.MonkeyPatch, + event: asyncio.Event, + event2: asyncio.Event, + mock: MagicMock, ) -> None: - error = ValueError("message processing failed") - received_errors: list[BaseException] = [] + error = RuntimeError("message processing failed") async def exception_handler(exc: BaseException) -> bool: - received_errors.append(exc) + mock(exc) + event.set() return True broker = self.get_broker() - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, + args, kwargs = self.get_subscriber_params( + queue, exception_handler=exception_handler, ) - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) - - await subscriber.consume(object()) - - assert received_errors == [error] - - async def test_subscriber_exception_handler_false_propagates_error( - self, - queue: str, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - error = ValueError("message processing failed") - received_errors: list[BaseException] = [] - - async def exception_handler(exc: BaseException) -> bool: - received_errors.append(exc) - return False + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + if msg == "hello": + raise error + event2.set() - broker = self.get_broker() - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, - exception_handler=exception_handler, - ) + async with self.patch_broker(broker) as br: + await br.start() - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) + await br.publish("hello", queue) + await asyncio.wait_for(event.wait(), timeout=self.timeout) - with pytest.raises( - ValueError, - match="message processing failed", - ) as exc_info: - await subscriber.consume(object()) + await br.publish("next", queue) + await asyncio.wait_for(event2.wait(), timeout=self.timeout) - assert exc_info.value is error - assert received_errors == [error] + mock.assert_called_once_with(error) + @pytest.mark.connected() async def test_subscriber_exception_handler_receives_context( self, queue: str, - monkeypatch: pytest.MonkeyPatch, + event: asyncio.Event, + event2: asyncio.Event, + mock: MagicMock, ) -> None: - error = ValueError("message processing failed") + error = RuntimeError("message processing failed") expected_service = object() - received: list[tuple[BaseException, object]] = [] async def exception_handler( exc: BaseException, service: object = Context("error_service"), ) -> bool: - received.append((exc, service)) + mock(exc, service) + event.set() return True broker = self.get_broker(apply_types=True) broker.context.set_global("error_service", expected_service) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, + args, kwargs = self.get_subscriber_params( + queue, exception_handler=exception_handler, ) - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + if msg == "hello": + raise error + event2.set() + + async with self.patch_broker(broker) as br: + await br.start() + + await br.publish("hello", queue) + await asyncio.wait_for(event.wait(), timeout=self.timeout) - await subscriber.consume(object()) + await br.publish("next", queue) + await asyncio.wait_for(event2.wait(), timeout=self.timeout) - assert received == [(error, expected_service)] + mock.assert_called_once_with(error, expected_service) + @pytest.mark.connected() async def test_subscriber_handles_error_without_calling_broker( self, queue: str, - monkeypatch: pytest.MonkeyPatch, + event: asyncio.Event, + event2: asyncio.Event, + mock: MagicMock, + mock2: MagicMock, ) -> None: - error = ValueError("message processing failed") - calls: list[tuple[str, BaseException]] = [] + error = RuntimeError("message processing failed") async def local_handler(exc: BaseException) -> bool: - calls.append(("subscriber", exc)) + mock(exc) + event.set() return True async def broker_handler(exc: BaseException) -> bool: - calls.append(("broker", exc)) + mock2(exc) return False broker = self.get_broker(exception_handler=broker_handler) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, + args, kwargs = self.get_subscriber_params( + queue, exception_handler=local_handler, ) - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) - - await subscriber.consume(object()) - - assert calls == [("subscriber", error)] - - async def test_broker_handles_error_after_subscriber_declines( - self, - queue: str, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - error = ValueError("message processing failed") - calls: list[str] = [] - - def local_handler(exc: BaseException) -> bool: - calls.append("subscriber") - return False - - def broker_handler(exc: BaseException) -> bool: - calls.append("broker") - return True + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + if msg == "hello": + raise error + event2.set() - broker = self.get_broker(exception_handler=broker_handler) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, - exception_handler=local_handler, - ) + async with self.patch_broker(broker) as br: + await br.start() - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) + await br.publish("hello", queue) + await asyncio.wait_for(event.wait(), timeout=self.timeout) - await subscriber.consume(object()) + await br.publish("next", queue) + await asyncio.wait_for(event2.wait(), timeout=self.timeout) - assert calls == ["subscriber", "broker"] + mock.assert_called_once_with(error) + mock2.assert_not_called() - async def test_error_propagates_when_both_handlers_decline( + @pytest.mark.connected() + async def test_broker_handles_error_after_subscriber_declines( self, queue: str, - monkeypatch: pytest.MonkeyPatch, + event: asyncio.Event, + event2: asyncio.Event, + mock: MagicMock, ) -> None: - error = ValueError("message processing failed") - calls: list[tuple[str, BaseException]] = [] + error = RuntimeError("message processing failed") async def local_handler(exc: BaseException) -> bool: - calls.append(("subscriber", exc)) + mock("subscriber", exc) return False async def broker_handler(exc: BaseException) -> bool: - calls.append(("broker", exc)) - return False + mock("broker", exc) + event.set() + return True broker = self.get_broker(exception_handler=broker_handler) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber( - *args, - **kwargs, + args, kwargs = self.get_subscriber_params( + queue, exception_handler=local_handler, ) - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) - - with pytest.raises( - ValueError, - match="message processing failed", - ) as exc_info: - await subscriber.consume(object()) - - assert exc_info.value is error - assert calls == [ - ("subscriber", error), - ("broker", error), - ] - - async def test_broker_exception_handler_false_propagates_error( - self, - queue: str, - monkeypatch: pytest.MonkeyPatch, - ) -> None: - error = ValueError("message processing failed") - received_errors: list[BaseException] = [] + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + if msg == "hello": + raise error + event2.set() - async def exception_handler(exc: BaseException) -> bool: - received_errors.append(exc) - return False + async with self.patch_broker(broker) as br: + await br.start() - broker = self.get_broker(exception_handler=exception_handler) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber(*args, **kwargs) - - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) + await br.publish("hello", queue) + await asyncio.wait_for(event.wait(), timeout=self.timeout) - with pytest.raises( - ValueError, - match="message processing failed", - ) as exc_info: - await subscriber.consume(object()) + await br.publish("next", queue) + await asyncio.wait_for(event2.wait(), timeout=self.timeout) - assert exc_info.value is error - assert received_errors == [error] + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] + @pytest.mark.connected() async def test_broker_exception_handler_true_suppresses_error( self, queue: str, - monkeypatch: pytest.MonkeyPatch, + event: asyncio.Event, + event2: asyncio.Event, + mock: MagicMock, ) -> None: - error = ValueError("message processing failed") - received_errors: list[BaseException] = [] + error = RuntimeError("message processing failed") async def exception_handler(exc: BaseException) -> bool: - received_errors.append(exc) + mock(exc) + event.set() return True broker = self.get_broker(exception_handler=exception_handler) args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber(*args, **kwargs) - - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) - await subscriber.consume(object()) + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + if msg == "hello": + raise error + event2.set() + + async with self.patch_broker(broker) as br: + await br.start() + + await br.publish("hello", queue) + await asyncio.wait_for(event.wait(), timeout=self.timeout) + + await br.publish("next", queue) + await asyncio.wait_for(event2.wait(), timeout=self.timeout) - assert received_errors == [error] + mock.assert_called_once_with(error) + @pytest.mark.connected() async def test_broker_exception_handler_receives_context( self, queue: str, - monkeypatch: pytest.MonkeyPatch, + event: asyncio.Event, + event2: asyncio.Event, + mock: MagicMock, ) -> None: - error = ValueError("message processing failed") + error = RuntimeError("message processing failed") expected_service = object() - received: list[tuple[BaseException, object]] = [] async def exception_handler( exc: BaseException, service: object = Context("error_service"), ) -> bool: - received.append((exc, service)) + mock(exc, service) + event.set() return True broker = self.get_broker( @@ -316,16 +242,20 @@ async def exception_handler( broker.context.set_global("error_service", expected_service) args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber(*args, **kwargs) - - subscriber._build_fastdepends_model() - monkeypatch.setattr(subscriber, "running", True) - monkeypatch.setattr( - subscriber, - "process_message", - AsyncMock(side_effect=error), - ) - await subscriber.consume(object()) + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + if msg == "hello": + raise error + event2.set() + + async with self.patch_broker(broker) as br: + await br.start() + + await br.publish("hello", queue) + await asyncio.wait_for(event.wait(), timeout=self.timeout) + + await br.publish("next", queue) + await asyncio.wait_for(event2.wait(), timeout=self.timeout) - assert received == [(error, expected_service)] + mock.assert_called_once_with(error, expected_service) diff --git a/tests/brokers/kafka/test_exception_handlers.py b/tests/brokers/kafka/test_exception_handlers.py index cef1e4e52c2..5fd1ef407a9 100644 --- a/tests/brokers/kafka/test_exception_handlers.py +++ b/tests/brokers/kafka/test_exception_handlers.py @@ -1,10 +1,139 @@ -import pytest - -from tests.brokers.base.exception_handlers import ExceptionHandlersTestcase - -from .basic import KafkaTestcaseConfig - - -@pytest.mark.kafka() -class TestExceptionHandlers(KafkaTestcaseConfig, ExceptionHandlersTestcase): - pass +import asyncio +from typing import Any +from unittest.mock import MagicMock, call + +import pytest + +from tests.brokers.base.exception_handlers import ExceptionHandlersTestcase + +from .basic import KafkaTestcaseConfig + + +@pytest.mark.kafka() +class TestExceptionHandlers(KafkaTestcaseConfig, ExceptionHandlersTestcase): + @pytest.mark.connected() + async def test_subscriber_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber # type: ignore[untyped-decorator] + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.start() + (consume_task,) = subscriber.tasks + + await br.publish("hello", queue) + + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await asyncio.wait_for( + consume_task, + timeout=self.timeout, + ) + + assert exc_info.value is error + mock.assert_called_once_with(error) + + @pytest.mark.connected() + async def test_broker_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber # type: ignore[untyped-decorator] + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.start() + (consume_task,) = subscriber.tasks + + await br.publish("hello", queue) + + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await asyncio.wait_for( + consume_task, + timeout=self.timeout, + ) + + assert exc_info.value is error + mock.assert_called_once_with(error) + + @pytest.mark.connected() + async def test_error_propagates_when_both_handlers_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber # type: ignore[untyped-decorator] + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + await br.start() + (consume_task,) = subscriber.tasks + + await br.publish("hello", queue) + + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await asyncio.wait_for( + consume_task, + timeout=self.timeout, + ) + + assert exc_info.value is error + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] From 94294f3921e8230b86f0a2200c13b7320d123cc4 Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Tue, 6 Oct 2026 23:33:16 +0300 Subject: [PATCH 5/7] feat: add application exception handler and refactor handler setup --- faststream/_internal/broker/broker.py | 13 +- faststream/_internal/configs/broker.py | 16 +- faststream/_internal/configs/endpoint.py | 13 +- .../_internal/endpoint/subscriber/usecase.py | 50 +- faststream/app.py | 11 + faststream/confluent/subscriber/config.py | 1 + faststream/kafka/broker/broker.py | 5 +- faststream/kafka/subscriber/config.py | 1 + faststream/kafka/subscriber/factory.py | 5 +- tests/brokers/base/exception_handlers.py | 637 +++++++++++++++--- .../brokers/kafka/test_exception_handlers.py | 147 +--- 11 files changed, 640 insertions(+), 259 deletions(-) diff --git a/faststream/_internal/broker/broker.py b/faststream/_internal/broker/broker.py index ab24de1dc51..cceef1b080d 100644 --- a/faststream/_internal/broker/broker.py +++ b/faststream/_internal/broker/broker.py @@ -13,7 +13,6 @@ ConnectionType, MsgType, ) -from faststream._internal.utils import apply_types, to_async from .pub_base import BrokerPublishMixin from .registrator import Registrator @@ -25,7 +24,6 @@ from faststream._internal.di import FastDependsConfig from faststream._internal.endpoint.subscriber import SubscriberUsecase from faststream._internal.producer import ProducerProto - from faststream._internal.types import AsyncExceptionHandler from faststream.specification.schema import BrokerSpec @@ -56,15 +54,8 @@ def __init__( routers=routers, config=config, ) - if config.broker_exception_handler is not None: - async_handler: AsyncExceptionHandler = to_async( - config.broker_exception_handler, - ) - config._broker_exception_handler = apply_types( - async_handler, - serializer_cls=config.fd_config._serializer, - context__=config.context, - ) + self._exception_handler = config.broker_exception_handler + self.specification = specification self.running = False diff --git a/faststream/_internal/configs/broker.py b/faststream/_internal/configs/broker.py index 9926380c052..0225fa087b4 100644 --- a/faststream/_internal/configs/broker.py +++ b/faststream/_internal/configs/broker.py @@ -9,6 +9,7 @@ from faststream._internal.logger import LoggerState from faststream._internal.producer import ProducerProto, ProducerUnset from faststream._internal.types import IdGenerator +from faststream._internal.utils import apply_types from faststream.message import gen_cor_id if TYPE_CHECKING: @@ -20,7 +21,6 @@ AsyncExceptionHandler, BrokerMiddleware, CustomCallable, - ExceptionHandler, ) from faststream.middlewares import AckPolicy @@ -34,12 +34,7 @@ class BrokerConfig: broker_parser: Optional["CustomCallable"] = None broker_decoder: Optional["CustomCallable"] = None broker_codec: Optional["CodecProto"] = None - broker_exception_handler: Optional["ExceptionHandler"] = None - _broker_exception_handler: Optional["AsyncExceptionHandler"] = field( - default=None, - init=False, - repr=False, - ) + broker_exception_handler: Optional["AsyncExceptionHandler"] = None producer: "ProducerProto[Any]" = field(default_factory=ProducerUnset) logger: "LoggerState" = field(default_factory=LoggerState) @@ -56,6 +51,13 @@ def __post_init__(self) -> None: # untyped callers still pass a generator: the first subscriber would spend it self.broker_dependencies = tuple(self.broker_dependencies) + if self.broker_exception_handler is not None: + self.broker_exception_handler = apply_types( + self.broker_exception_handler, + serializer_cls=self.fd_config._serializer, + context__=self.context, + ) + def __repr__(self) -> str: return f"{self.__class__.__name__}(id: {id(self)})" diff --git a/faststream/_internal/configs/endpoint.py b/faststream/_internal/configs/endpoint.py index 7ac7cd4a983..09f487d5db1 100644 --- a/faststream/_internal/configs/endpoint.py +++ b/faststream/_internal/configs/endpoint.py @@ -2,10 +2,11 @@ from typing import TYPE_CHECKING from faststream._internal.constants import EMPTY +from faststream._internal.utils import apply_types from faststream.middlewares import AckPolicy if TYPE_CHECKING: - from faststream._internal.types import AsyncCallable, ExceptionHandler + from faststream._internal.types import AsyncCallable, AsyncExceptionHandler from .broker import BrokerConfig @@ -23,13 +24,21 @@ class PublisherUsecaseConfig(EndpointConfig): @dataclass(kw_only=True) class SubscriberUsecaseConfig(EndpointConfig): no_reply: bool = False - exception_handler: "ExceptionHandler | None" = None + exception_handler: "AsyncExceptionHandler | None" = None _ack_policy: AckPolicy = field(default_factory=lambda: EMPTY, repr=False) parser: "AsyncCallable" = field(init=False) decoder: "AsyncCallable" = field(init=False) + def __post_init__(self) -> None: + if self.exception_handler is not None: + self.exception_handler = apply_types( + self.exception_handler, + serializer_cls=self._outer_config.fd_config._serializer, + context__=self._outer_config.context, + ) + @property def auto_ack_disabled(self) -> bool: return self.ack_policy is AckPolicy.MANUAL diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index 9a135608825..41bc2febef1 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -1,5 +1,5 @@ from abc import abstractmethod -from collections.abc import AsyncIterator, Callable, Iterable, Sequence +from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence from contextlib import AbstractContextManager, AsyncExitStack, nullcontext from itertools import chain from types import TracebackType @@ -24,7 +24,6 @@ P_HandlerParams, T_HandlerReturn, ) -from faststream._internal.utils import apply_types from faststream._internal.utils.functions import to_async from faststream.exceptions import StopConsume, SubscriberNotFound from faststream.middlewares import AcknowledgementMiddleware @@ -104,16 +103,7 @@ def __init__( self._no_reply = config.no_reply self._parser = config.parser self._decoder = config.decoder - self._exception_handler_call: AsyncExceptionHandler | None = None - if config.exception_handler is not None: - async_handler: AsyncExceptionHandler = to_async( - config.exception_handler, - ) - self._exception_handler_call = apply_types( - async_handler, - serializer_cls=self._outer_config.fd_config._serializer, - context__=self._outer_config.context, - ) + self._exception_handler = config.exception_handler self.ack_policy = config.ack_policy self.__auto_ack_disabled = config.auto_ack_disabled @@ -340,16 +330,35 @@ def real_wrapper( return real_wrapper(func) + @property + def _app_exception_handler(self) -> "AsyncExceptionHandler | None": + app_handler: AsyncExceptionHandler | None = getattr( + self._outer_config.context.get("app"), + "_exception_handler", + None, + ) + return app_handler + async def _handle_exception(self, exc: BaseException) -> bool: - local_handler = self._exception_handler_call - if local_handler is not None and await local_handler(exc): + context = self._outer_config.context + + local_handler = cast( + "Callable[..., Awaitable[bool]] | None", + self._exception_handler, + ) + if local_handler is not None and await local_handler(exc, context__=context): return True - broker_handler: AsyncExceptionHandler | None = ( - self._outer_config._broker_exception_handler + broker_handler = cast( + "Callable[..., Awaitable[bool]] | None", + self._outer_config.broker_exception_handler, ) - if broker_handler is not None: - return await broker_handler(exc) + if broker_handler is not None and await broker_handler(exc, context__=context): + return True + + app_handler = self._app_exception_handler + if app_handler is not None: + return await app_handler(exc) return False @@ -375,8 +384,9 @@ async def consume(self, msg: MsgType) -> Any: except Exception as exc: # All other exceptions were logged by CriticalLogMiddleware has_handler = ( - self._exception_handler_call is not None - or self._outer_config._broker_exception_handler is not None + self._exception_handler is not None + or self._outer_config.broker_exception_handler is not None + or self._app_exception_handler is not None ) if has_handler and not await self._handle_exception(exc): raise diff --git a/faststream/app.py b/faststream/app.py index cea33f42d95..b21f916d9a0 100644 --- a/faststream/app.py +++ b/faststream/app.py @@ -19,6 +19,7 @@ from faststream._internal.context import ContextRepo from faststream._internal.di import FastDependsConfig from faststream._internal.logger import logger +from faststream._internal.utils import apply_types, to_async from faststream.asgi.app import AsgiFastStream if TYPE_CHECKING: @@ -31,6 +32,7 @@ SettingField, ) from faststream._internal.broker import BrokerUsecase + from faststream._internal.types import AsyncExceptionHandler, ExceptionHandler from faststream.asgi import AsyncAPIRoute from faststream.asgi.types import ASGIApp from faststream.specification.base import SpecificationFactory @@ -49,6 +51,7 @@ def __init__( provider: Optional["Provider"] = None, serializer: Optional["SerializerProto"] = EMPTY, context: ContextRepo | None = None, + exception_handler: "ExceptionHandler | None" = None, lifespan: Optional["Lifespan"] = None, on_startup: Sequence["AnyCallable"] = (), after_startup: Sequence["AnyCallable"] = (), @@ -72,6 +75,14 @@ def __init__( specification=specification, ) + self._exception_handler: AsyncExceptionHandler | None = None + if exception_handler is not None: + self._exception_handler = apply_types( + to_async(exception_handler), + serializer_cls=self.config._serializer, + context__=self.context, + ) + self._should_exit = False async def run( diff --git a/faststream/confluent/subscriber/config.py b/faststream/confluent/subscriber/config.py index 729c65c90b1..ae3f9f8ca11 100644 --- a/faststream/confluent/subscriber/config.py +++ b/faststream/confluent/subscriber/config.py @@ -31,6 +31,7 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): connection_data: dict[str, Any] = field(default_factory=dict) def __post_init__(self) -> None: + super().__post_init__() self.connection_data["enable_auto_commit"] = self.ack_first @property diff --git a/faststream/kafka/broker/broker.py b/faststream/kafka/broker/broker.py index 788dca6399d..01615bb6368 100644 --- a/faststream/kafka/broker/broker.py +++ b/faststream/kafka/broker/broker.py @@ -26,6 +26,7 @@ from faststream._internal.context.repository import ContextRepo from faststream._internal.di import FastDependsConfig from faststream._internal.types import IdGenerator +from faststream._internal.utils import to_async from faststream._internal.utils.data import filter_by_dict from faststream.exceptions import IncorrectState from faststream.kafka._compat import AIOKAFKA_V013, validate_client_rack @@ -465,7 +466,9 @@ def __init__( broker_decoder=decoder, broker_codec=codec, broker_parser=parser, - broker_exception_handler=exception_handler, + broker_exception_handler=( + to_async(exception_handler) if exception_handler is not None else None + ), broker_middlewares=middlewares, logger=make_kafka_logger_state( logger=logger, diff --git a/faststream/kafka/subscriber/config.py b/faststream/kafka/subscriber/config.py index 8c5eee0f73a..f276485237e 100644 --- a/faststream/kafka/subscriber/config.py +++ b/faststream/kafka/subscriber/config.py @@ -35,6 +35,7 @@ class KafkaSubscriberConfig(SubscriberUsecaseConfig): partitions: Iterable["TopicPartition"] = field(default_factory=list) def __post_init__(self) -> None: + super().__post_init__() self.connection_args["enable_auto_commit"] = self.ack_first @property diff --git a/faststream/kafka/subscriber/factory.py b/faststream/kafka/subscriber/factory.py index ccd66281542..93916f6a644 100644 --- a/faststream/kafka/subscriber/factory.py +++ b/faststream/kafka/subscriber/factory.py @@ -4,6 +4,7 @@ from faststream._internal.constants import EMPTY from faststream._internal.endpoint.subscriber.call_item import CallsCollection +from faststream._internal.utils import to_async from faststream.exceptions import SetupError from faststream.kafka._compat import validate_client_rack from faststream.middlewares import AckPolicy @@ -70,7 +71,9 @@ def create_subscriber( listener=listener, pattern=pattern, no_reply=no_reply, - exception_handler=exception_handler, + exception_handler=( + to_async(exception_handler) if exception_handler is not None else None + ), _outer_config=config, _ack_policy=ack_policy, ) diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py index bca4abaf4b3..8e882448bbb 100644 --- a/tests/brokers/base/exception_handlers.py +++ b/tests/brokers/base/exception_handlers.py @@ -1,28 +1,32 @@ -import asyncio from typing import Any from unittest.mock import MagicMock, call import pytest -from faststream import Context +from faststream import Context, FastStream, TestApp from tests.brokers.base.basic import BaseTestcaseConfig @pytest.mark.asyncio() class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): - @pytest.mark.connected() + async def get_consume_message( + self, + broker: Any, + body: Any, + queue: str, + ) -> Any: + """Build a raw message; Kafka overrides this with a ConsumerRecord.""" + raise NotImplementedError + async def test_subscriber_exception_handler_true_suppresses_error( self, queue: str, - event: asyncio.Event, - event2: asyncio.Event, mock: MagicMock, ) -> None: error = RuntimeError("message processing failed") async def exception_handler(exc: BaseException) -> bool: mock(exc) - event.set() return True broker = self.get_broker() @@ -30,30 +34,21 @@ async def exception_handler(exc: BaseException) -> bool: queue, exception_handler=exception_handler, ) + subscriber = broker.subscriber(*args, **kwargs) - @broker.subscriber(*args, **kwargs) + @subscriber async def handler(msg: Any) -> None: - if msg == "hello": - raise error - event2.set() + raise error async with self.patch_broker(broker) as br: - await br.start() - - await br.publish("hello", queue) - await asyncio.wait_for(event.wait(), timeout=self.timeout) - - await br.publish("next", queue) - await asyncio.wait_for(event2.wait(), timeout=self.timeout) + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) mock.assert_called_once_with(error) - @pytest.mark.connected() async def test_subscriber_exception_handler_receives_context( self, queue: str, - event: asyncio.Event, - event2: asyncio.Event, mock: MagicMock, ) -> None: error = RuntimeError("message processing failed") @@ -64,7 +59,6 @@ async def exception_handler( service: object = Context("error_service"), ) -> bool: mock(exc, service) - event.set() return True broker = self.get_broker(apply_types=True) @@ -74,30 +68,21 @@ async def exception_handler( queue, exception_handler=exception_handler, ) + subscriber = broker.subscriber(*args, **kwargs) - @broker.subscriber(*args, **kwargs) + @subscriber async def handler(msg: Any) -> None: - if msg == "hello": - raise error - event2.set() + raise error async with self.patch_broker(broker) as br: - await br.start() - - await br.publish("hello", queue) - await asyncio.wait_for(event.wait(), timeout=self.timeout) - - await br.publish("next", queue) - await asyncio.wait_for(event2.wait(), timeout=self.timeout) + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) mock.assert_called_once_with(error, expected_service) - @pytest.mark.connected() async def test_subscriber_handles_error_without_calling_broker( self, queue: str, - event: asyncio.Event, - event2: asyncio.Event, mock: MagicMock, mock2: MagicMock, ) -> None: @@ -105,7 +90,6 @@ async def test_subscriber_handles_error_without_calling_broker( async def local_handler(exc: BaseException) -> bool: mock(exc) - event.set() return True async def broker_handler(exc: BaseException) -> bool: @@ -117,31 +101,22 @@ async def broker_handler(exc: BaseException) -> bool: queue, exception_handler=local_handler, ) + subscriber = broker.subscriber(*args, **kwargs) - @broker.subscriber(*args, **kwargs) + @subscriber async def handler(msg: Any) -> None: - if msg == "hello": - raise error - event2.set() + raise error async with self.patch_broker(broker) as br: - await br.start() - - await br.publish("hello", queue) - await asyncio.wait_for(event.wait(), timeout=self.timeout) - - await br.publish("next", queue) - await asyncio.wait_for(event2.wait(), timeout=self.timeout) + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) mock.assert_called_once_with(error) mock2.assert_not_called() - @pytest.mark.connected() async def test_broker_handles_error_after_subscriber_declines( self, queue: str, - event: asyncio.Event, - event2: asyncio.Event, mock: MagicMock, ) -> None: error = RuntimeError("message processing failed") @@ -152,7 +127,6 @@ async def local_handler(exc: BaseException) -> bool: async def broker_handler(exc: BaseException) -> bool: mock("broker", exc) - event.set() return True broker = self.get_broker(exception_handler=broker_handler) @@ -160,68 +134,49 @@ async def broker_handler(exc: BaseException) -> bool: queue, exception_handler=local_handler, ) + subscriber = broker.subscriber(*args, **kwargs) - @broker.subscriber(*args, **kwargs) + @subscriber async def handler(msg: Any) -> None: - if msg == "hello": - raise error - event2.set() + raise error async with self.patch_broker(broker) as br: - await br.start() - - await br.publish("hello", queue) - await asyncio.wait_for(event.wait(), timeout=self.timeout) - - await br.publish("next", queue) - await asyncio.wait_for(event2.wait(), timeout=self.timeout) + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) assert mock.call_args_list == [ call("subscriber", error), call("broker", error), ] - @pytest.mark.connected() async def test_broker_exception_handler_true_suppresses_error( self, queue: str, - event: asyncio.Event, - event2: asyncio.Event, mock: MagicMock, ) -> None: error = RuntimeError("message processing failed") async def exception_handler(exc: BaseException) -> bool: mock(exc) - event.set() return True broker = self.get_broker(exception_handler=exception_handler) args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) - @broker.subscriber(*args, **kwargs) + @subscriber async def handler(msg: Any) -> None: - if msg == "hello": - raise error - event2.set() + raise error async with self.patch_broker(broker) as br: - await br.start() - - await br.publish("hello", queue) - await asyncio.wait_for(event.wait(), timeout=self.timeout) - - await br.publish("next", queue) - await asyncio.wait_for(event2.wait(), timeout=self.timeout) + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) mock.assert_called_once_with(error) - @pytest.mark.connected() async def test_broker_exception_handler_receives_context( self, queue: str, - event: asyncio.Event, - event2: asyncio.Event, mock: MagicMock, ) -> None: error = RuntimeError("message processing failed") @@ -232,7 +187,6 @@ async def exception_handler( service: object = Context("error_service"), ) -> bool: mock(exc, service) - event.set() return True broker = self.get_broker( @@ -242,20 +196,523 @@ async def exception_handler( broker.context.set_global("error_service", expected_service) args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) + + mock.assert_called_once_with(error, expected_service) + + async def test_subscriber_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + message = await self.get_consume_message(br, "hello", queue) + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(message) + + mock.assert_called_once_with(error) + assert exc_info.value is error + + async def test_error_propagates_when_both_handlers_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) - @broker.subscriber(*args, **kwargs) + @subscriber async def handler(msg: Any) -> None: - if msg == "hello": - raise error - event2.set() + raise error async with self.patch_broker(broker) as br: - await br.start() + message = await self.get_consume_message(br, "hello", queue) + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(message) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] + assert exc_info.value is error + + async def test_broker_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + async with self.patch_broker(broker) as br: + message = await self.get_consume_message(br, "hello", queue) + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(message) + + mock.assert_called_once_with(error) + assert exc_info.value is error + + async def test_broker_exception_handlers_are_independent( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error1 = RuntimeError("first broker processing failed") + error2 = RuntimeError("second broker processing failed") + + async def exception_handler1(exc: BaseException) -> bool: + mock(exc) + return True + + async def exception_handler2(exc: BaseException) -> bool: + mock2(exc) + return True + + broker1 = self.get_broker(exception_handler=exception_handler1) + broker2 = self.get_broker(exception_handler=exception_handler2) + + args, kwargs = self.get_subscriber_params(queue) + args2, kwargs2 = self.get_subscriber_params(queue + "1") + subscriber1 = broker1.subscriber(*args, **kwargs) + subscriber2 = broker2.subscriber(*args2, **kwargs2) + + @subscriber1 + async def handler1(msg: Any) -> None: + raise error1 + + @subscriber2 + async def handler2(msg: Any) -> None: + raise error2 + + app = FastStream(broker1, broker2) + + async with ( + self.patch_broker(broker1, connect_only=True) as br1, + self.patch_broker(broker2, connect_only=True) as br2, + TestApp(app), + ): + message1 = await self.get_consume_message(br1, "hello", queue) + await subscriber1.consume(message1) + + mock2.assert_not_called() + + message2 = await self.get_consume_message(br2, "hello", queue + "1") + await subscriber2.consume(message2) + + mock.assert_called_once_with(error1) + mock2.assert_called_once_with(error2) + + @pytest.mark.parametrize("sync_handler", (False, True)) + async def test_app_exception_handler_true_suppresses_error( + self, + queue: str, + mock: MagicMock, + sync_handler: bool, + ) -> None: + error = RuntimeError("message processing failed") + + def sync_exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + async def async_exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream( + broker, + exception_handler=( + sync_exception_handler if sync_handler else async_exception_handler + ), + ) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) + + mock.assert_called_once_with(error) + + async def test_app_exception_handler_false_propagates_error( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return False + + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=exception_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(message) + + mock.assert_called_once_with(error) + assert exc_info.value is error + + async def test_app_handles_error_after_subscriber_and_broker_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + async def app_handler(exc: BaseException) -> bool: + mock("app", exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + call("app", error), + ] + + @pytest.mark.parametrize("handled_by", ("subscriber", "broker")) + async def test_app_handler_is_not_called_after_error_is_handled( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + handled_by: str, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return handled_by == "subscriber" + + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return True + + async def app_handler(exc: BaseException) -> bool: + mock2(exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) + + expected_calls = [call("subscriber", error)] + if handled_by == "broker": + expected_calls.append(call("broker", error)) + + assert mock.call_args_list == expected_calls + mock2.assert_not_called() + + async def test_error_propagates_when_all_three_handlers_decline( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False + + def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return False + + def app_handler(exc: BaseException) -> bool: + mock("app", exc) + return False + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await subscriber.consume(message) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + call("app", error), + ] + assert exc_info.value is error + + async def test_app_exception_handler_receives_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=exception_handler) + app.context.set_global("error_service", expected_service) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) + + mock.assert_called_once_with(error, expected_service) + + async def test_app_exception_handlers_are_independent( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error1 = RuntimeError("first app processing failed") + error2 = RuntimeError("second app processing failed") + + async def exception_handler1(exc: BaseException) -> bool: + mock(exc) + return True + + async def exception_handler2(exc: BaseException) -> bool: + mock2(exc) + return True + + broker1, broker2 = self.get_broker(), self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + args2, kwargs2 = self.get_subscriber_params(queue + "1") + subscriber1 = broker1.subscriber(*args, **kwargs) + subscriber2 = broker2.subscriber(*args2, **kwargs2) + + @subscriber1 + async def handler1(msg: Any) -> None: + raise error1 + + @subscriber2 + async def handler2(msg: Any) -> None: + raise error2 + + app1 = FastStream(broker1, exception_handler=exception_handler1) + app2 = FastStream(broker2, exception_handler=exception_handler2) + + async with ( + self.patch_broker(broker1, connect_only=True) as br1, + self.patch_broker(broker2, connect_only=True) as br2, + TestApp(app1), + TestApp(app2), + ): + message1 = await self.get_consume_message(br1, "hello", queue) + await subscriber1.consume(message1) + mock2.assert_not_called() + + message2 = await self.get_consume_message(br2, "hello", queue + "1") + await subscriber2.consume(message2) + + mock.assert_called_once_with(error1) + mock2.assert_called_once_with(error2) + + async def test_app_without_exception_handlers_preserves_error_suppression( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + broker = self.get_broker() + args, kwargs = self.get_subscriber_params(queue) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + mock(msg) + raise error + + app = FastStream(broker) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) + + mock.assert_called_once_with("hello") + + @pytest.mark.parametrize("handler_level", ("subscriber", "broker")) + async def test_exception_handler_receives_dependency_from_app_context( + self, + queue: str, + mock: MagicMock, + handler_level: str, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + if handler_level == "broker": + broker = self.get_broker( + apply_types=True, + exception_handler=exception_handler, + ) + args, kwargs = self.get_subscriber_params(queue) + else: + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error - await br.publish("hello", queue) - await asyncio.wait_for(event.wait(), timeout=self.timeout) + app = FastStream(broker) + app.context.set_global("error_service", expected_service) - await br.publish("next", queue) - await asyncio.wait_for(event2.wait(), timeout=self.timeout) + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + message = await self.get_consume_message(br, "hello", queue) + await subscriber.consume(message) mock.assert_called_once_with(error, expected_service) diff --git a/tests/brokers/kafka/test_exception_handlers.py b/tests/brokers/kafka/test_exception_handlers.py index 5fd1ef407a9..3571a099f03 100644 --- a/tests/brokers/kafka/test_exception_handlers.py +++ b/tests/brokers/kafka/test_exception_handlers.py @@ -1,139 +1,32 @@ -import asyncio -from typing import Any -from unittest.mock import MagicMock, call +from typing import TYPE_CHECKING, Any import pytest +from typing_extensions import override +from faststream.kafka.testing import build_message from tests.brokers.base.exception_handlers import ExceptionHandlersTestcase -from .basic import KafkaTestcaseConfig +from .basic import KafkaMemoryTestcaseConfig +if TYPE_CHECKING: + from aiokafka import ConsumerRecord -@pytest.mark.kafka() -class TestExceptionHandlers(KafkaTestcaseConfig, ExceptionHandlersTestcase): - @pytest.mark.connected() - async def test_subscriber_exception_handler_false_propagates_error( - self, - queue: str, - mock: MagicMock, - ) -> None: - error = RuntimeError("message processing failed") - - async def exception_handler(exc: BaseException) -> bool: - mock(exc) - return False - - broker = self.get_broker() - args, kwargs = self.get_subscriber_params( - queue, - exception_handler=exception_handler, - ) - subscriber = broker.subscriber(*args, **kwargs) - - @subscriber # type: ignore[untyped-decorator] - async def handler(msg: Any) -> None: - raise error - - async with self.patch_broker(broker) as br: - await br.start() - (consume_task,) = subscriber.tasks - - await br.publish("hello", queue) - - with pytest.raises( - RuntimeError, - match="message processing failed", - ) as exc_info: - await asyncio.wait_for( - consume_task, - timeout=self.timeout, - ) - - assert exc_info.value is error - mock.assert_called_once_with(error) - - @pytest.mark.connected() - async def test_broker_exception_handler_false_propagates_error( - self, - queue: str, - mock: MagicMock, - ) -> None: - error = RuntimeError("message processing failed") - - async def exception_handler(exc: BaseException) -> bool: - mock(exc) - return False + from faststream.kafka import KafkaBroker - broker = self.get_broker(exception_handler=exception_handler) - args, kwargs = self.get_subscriber_params(queue) - subscriber = broker.subscriber(*args, **kwargs) - @subscriber # type: ignore[untyped-decorator] - async def handler(msg: Any) -> None: - raise error - - async with self.patch_broker(broker) as br: - await br.start() - (consume_task,) = subscriber.tasks - - await br.publish("hello", queue) - - with pytest.raises( - RuntimeError, - match="message processing failed", - ) as exc_info: - await asyncio.wait_for( - consume_task, - timeout=self.timeout, - ) - - assert exc_info.value is error - mock.assert_called_once_with(error) - - @pytest.mark.connected() - async def test_error_propagates_when_both_handlers_decline( +@pytest.mark.kafka() +class TestExceptionHandlers(KafkaMemoryTestcaseConfig, ExceptionHandlersTestcase): + @override + async def get_consume_message( self, + broker: "KafkaBroker", + body: Any, queue: str, - mock: MagicMock, - ) -> None: - error = RuntimeError("message processing failed") - - def local_handler(exc: BaseException) -> bool: - mock("subscriber", exc) - return False - - def broker_handler(exc: BaseException) -> bool: - mock("broker", exc) - return False - - broker = self.get_broker(exception_handler=broker_handler) - args, kwargs = self.get_subscriber_params( - queue, - exception_handler=local_handler, + ) -> "ConsumerRecord": + return await build_message( + message=body, + topic=queue, + serializer=broker.config.fd_config._serializer, + codec=broker.config.broker_codec, + id_generator=broker.config.id_generator, ) - subscriber = broker.subscriber(*args, **kwargs) - - @subscriber # type: ignore[untyped-decorator] - async def handler(msg: Any) -> None: - raise error - - async with self.patch_broker(broker) as br: - await br.start() - (consume_task,) = subscriber.tasks - - await br.publish("hello", queue) - - with pytest.raises( - RuntimeError, - match="message processing failed", - ) as exc_info: - await asyncio.wait_for( - consume_task, - timeout=self.timeout, - ) - - assert exc_info.value is error - assert mock.call_args_list == [ - call("subscriber", error), - call("broker", error), - ] From 85210f4b771e99ae8186307ce38b27aa66ff0a72 Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Tue, 13 Oct 2026 21:19:34 +0300 Subject: [PATCH 6/7] fix: handle processing errors and document exception handlers --- .../exception-handlers/index.md | 61 ++++++ docs/docs/navigation_template.txt | 1 + .../exception_handlers/__init__.py | 0 .../exception_handlers/kafka.py | 37 ++++ .../_internal/endpoint/subscriber/usecase.py | 43 ++--- faststream/_internal/types.py | 4 +- faststream/_internal/utils/__init__.py | 3 +- faststream/_internal/utils/functions.py | 14 ++ faststream/app.py | 6 +- faststream/kafka/broker/broker.py | 6 +- faststream/kafka/subscriber/factory.py | 6 +- tests/brokers/base/exception_handlers.py | 177 +++++++++++++----- .../brokers/kafka/test_exception_handlers.py | 25 +-- .../exception_handlers/__init__.py | 1 + .../exception_handlers/test_handlers.py | 32 ++++ 15 files changed, 302 insertions(+), 114 deletions(-) create mode 100644 docs/docs/en/getting-started/exception-handlers/index.md create mode 100644 docs/docs_src/getting_started/exception_handlers/__init__.py create mode 100644 docs/docs_src/getting_started/exception_handlers/kafka.py create mode 100644 tests/docs/getting_started/exception_handlers/__init__.py create mode 100644 tests/docs/getting_started/exception_handlers/test_handlers.py diff --git a/docs/docs/en/getting-started/exception-handlers/index.md b/docs/docs/en/getting-started/exception-handlers/index.md new file mode 100644 index 00000000000..bf74037bf45 --- /dev/null +++ b/docs/docs/en/getting-started/exception-handlers/index.md @@ -0,0 +1,61 @@ +--- +description: >- + Handle AIOKafka message processing errors with subscriber, broker, + and FastStream exception handlers. +--- + +# Exception Handlers + +With AIOKafka, you can register an `exception_handler` on a subscriber, +on the broker, or on the `FastStream` application. + +Handlers can be synchronous or asynchronous. Each handler receives the +exception and returns a boolean. + +## Handler Order + +FastStream calls the handlers in this order: + +1. The subscriber's handler. +2. The broker's handler. +3. The application's handler. + +Missing handlers are skipped. Processing stops when a handler returns `True`. + +## Return Value + +- `True`: the exception is handled and suppressed. +- `False`: processing continues with the next handler. + +If no handler returns `True`, the original exception is raised again. + +## Example + +```python linenums="1" +{!> docs_src/getting_started/exception_handlers/kafka.py !} +``` + +Publish one of these messages to the `exception-handlers` topic: + +| Message | Exception | Result | +|---------|-----------|--------| +| `subscriber` | `ValueError` | Suppressed by the subscriber's handler | +| `broker` | `RuntimeError` | Suppressed by the broker's handler | +| `application` | `TypeError` | Logged by the application's handler and raised again | + +## Context + +Exception handlers support [context injection](../context.md){.internal-link}. +The application handler in the example uses `Logger` to access the logger +for the current message. + +## Middleware and Acknowledgement + +If middleware has already suppressed an exception, the exception handlers +are not called. + +Returning `True` controls exception propagation. Message acknowledgement +follows the broker's [acknowledgement policy](../acknowledgement.md){.internal-link}. + +`StopConsume` and `StopApplication` continue through the normal shutdown +handling and are not passed to these callbacks. \ No newline at end of file diff --git a/docs/docs/navigation_template.txt b/docs/docs/navigation_template.txt index 84ca1491602..e796e0c57bc 100644 --- a/docs/docs/navigation_template.txt +++ b/docs/docs/navigation_template.txt @@ -22,6 +22,7 @@ search: - [Routers](getting-started/routers/index.md) - [Dependencies](getting-started/dependencies/index.md) - [Context](getting-started/context.md) + - [Exception Handlers](getting-started/exception-handlers/index.md) - [Custom Serialization](getting-started/serialization/index.md) - [Parser](getting-started/serialization/parser.md) - [Decoder](getting-started/serialization/decoder.md) diff --git a/docs/docs_src/getting_started/exception_handlers/__init__.py b/docs/docs_src/getting_started/exception_handlers/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/docs/docs_src/getting_started/exception_handlers/kafka.py b/docs/docs_src/getting_started/exception_handlers/kafka.py new file mode 100644 index 00000000000..b31c19e3a8f --- /dev/null +++ b/docs/docs_src/getting_started/exception_handlers/kafka.py @@ -0,0 +1,37 @@ +from faststream import FastStream, Logger +from faststream.kafka import KafkaBroker + + +def subscriber_exception_handler(exc: BaseException) -> bool: + return isinstance(exc, ValueError) + + +async def broker_exception_handler(exc: BaseException) -> bool: + return isinstance(exc, RuntimeError) + + +async def app_exception_handler( + exc: BaseException, + logger: Logger, +) -> bool: + logger.error("Unhandled message processing error", exc_info=exc) + return False + + +broker = KafkaBroker( + "localhost:9092", + exception_handler=broker_exception_handler, +) +app = FastStream(broker, exception_handler=app_exception_handler) + + +@broker.subscriber( + "exception-handlers", + exception_handler=subscriber_exception_handler, +) +async def handle(message: str) -> None: + if message == "subscriber": + raise ValueError("subscriber error") + if message == "broker": + raise RuntimeError("broker error") + raise TypeError("application error") diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index d198beae9fe..ac585ce957a 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -1,5 +1,5 @@ from abc import abstractmethod -from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence +from collections.abc import AsyncIterator, Callable, Iterable, Sequence from contextlib import AbstractContextManager, AsyncExitStack, ExitStack, nullcontext from itertools import chain from types import TracebackType @@ -339,26 +339,24 @@ def _app_exception_handler(self) -> "AsyncExceptionHandler | None": ) return app_handler - async def _handle_exception(self, exc: BaseException) -> bool: + async def _handle_exception( + self, + _exc_type: type[BaseException] | None, + exc: BaseException | None, + _exc_tb: TracebackType | None, + ) -> bool: + if not isinstance(exc, Exception) or isinstance(exc, (StopConsume, SystemExit)): + return False + context = self._outer_config.context - local_handler = cast( - "Callable[..., Awaitable[bool]] | None", + for handler in ( self._exception_handler, - ) - if local_handler is not None and await local_handler(exc, context__=context): - return True - - broker_handler = cast( - "Callable[..., Awaitable[bool]] | None", self._outer_config.broker_exception_handler, - ) - if broker_handler is not None and await broker_handler(exc, context__=context): - return True - - app_handler = self._app_exception_handler - if app_handler is not None: - return await app_handler(exc) + self._app_exception_handler, + ): + if handler is not None and await handler(exc, context__=context): + return True return False @@ -381,16 +379,6 @@ async def consume(self, msg: MsgType) -> Any: if app := self._outer_config.context.get("app"): app.exit() - except Exception as exc: - # All other exceptions were logged by CriticalLogMiddleware - has_handler = ( - self._exception_handler is not None - or self._outer_config.broker_exception_handler is not None - or self._app_exception_handler is not None - ) - if has_handler and not await self._handle_exception(exc): - raise - async def process_message(self, msg: MsgType) -> "Response": """Execute all message processing stages.""" context = self._outer_config.context @@ -413,6 +401,7 @@ async def process_message(self, msg: MsgType) -> "Response": # reserve place for context scope under the middlewares __aexit__ # because middlewares should be exited before context scope release message_scope = stack.enter_context(ExitStack()) + stack.push_async_exit(self._handle_exception) # enter all middlewares middlewares: list[BaseMiddleware] = [] diff --git a/faststream/_internal/types.py b/faststream/_internal/types.py index c89a992fb05..379897c23e9 100644 --- a/faststream/_internal/types.py +++ b/faststream/_internal/types.py @@ -43,8 +43,8 @@ AsyncFilter: TypeAlias = Callable[[StreamMsg], Awaitable[bool]] Filter: TypeAlias = SyncFilter[StreamMsg] | AsyncFilter[StreamMsg] -SyncExceptionHandler: TypeAlias = Callable[[BaseException], bool] -AsyncExceptionHandler: TypeAlias = Callable[[BaseException], Awaitable[bool]] +SyncExceptionHandler: TypeAlias = Callable[..., bool] +AsyncExceptionHandler: TypeAlias = Callable[..., Awaitable[bool]] ExceptionHandler: TypeAlias = SyncExceptionHandler | AsyncExceptionHandler SyncCallable: TypeAlias = Callable[ diff --git a/faststream/_internal/utils/__init__.py b/faststream/_internal/utils/__init__.py index 061216397e1..2948d7cced7 100644 --- a/faststream/_internal/utils/__init__.py +++ b/faststream/_internal/utils/__init__.py @@ -1,8 +1,9 @@ from fast_depends import inject as apply_types -from .functions import to_async +from .functions import ensure_exception_handler, to_async __all__ = ( "apply_types", + "ensure_exception_handler", "to_async", ) diff --git a/faststream/_internal/utils/functions.py b/faststream/_internal/utils/functions.py index 97b92e4f1b7..09e5d40cde3 100644 --- a/faststream/_internal/utils/functions.py +++ b/faststream/_internal/utils/functions.py @@ -3,6 +3,7 @@ from concurrent.futures import Executor from functools import partial, wraps from typing import ( + TYPE_CHECKING, Any, TypeVar, cast, @@ -18,8 +19,12 @@ from faststream._internal.basic_types import F_Return, F_Spec +if TYPE_CHECKING: + from faststream._internal.types import AsyncExceptionHandler, ExceptionHandler + __all__ = ( "call_or_await", + "ensure_exception_handler", "to_async", ) @@ -56,6 +61,15 @@ async def to_async_wrapper(*args: F_Spec.args, **kwargs: F_Spec.kwargs) -> F_Ret return to_async_wrapper +def ensure_exception_handler( + handler: "ExceptionHandler | None", +) -> "AsyncExceptionHandler | None": + if handler is None: + return None + + return to_async(handler) + + async def return_input(x: Any) -> Any: return x diff --git a/faststream/app.py b/faststream/app.py index b21f916d9a0..86c4b109329 100644 --- a/faststream/app.py +++ b/faststream/app.py @@ -75,13 +75,13 @@ def __init__( specification=specification, ) - self._exception_handler: AsyncExceptionHandler | None = None if exception_handler is not None: - self._exception_handler = apply_types( + self._exception_handler: AsyncExceptionHandler | None = apply_types( to_async(exception_handler), serializer_cls=self.config._serializer, - context__=self.context, ) + else: + self._exception_handler = None self._should_exit = False diff --git a/faststream/kafka/broker/broker.py b/faststream/kafka/broker/broker.py index 8075803a805..329ae4d61a2 100644 --- a/faststream/kafka/broker/broker.py +++ b/faststream/kafka/broker/broker.py @@ -26,7 +26,7 @@ from faststream._internal.context.repository import ContextRepo from faststream._internal.di import FastDependsConfig from faststream._internal.types import IdGenerator -from faststream._internal.utils import to_async +from faststream._internal.utils import ensure_exception_handler from faststream._internal.utils.data import filter_by_dict from faststream.exceptions import IncorrectState from faststream.kafka._compat import AIOKAFKA_V013, validate_client_rack @@ -475,9 +475,7 @@ def __init__( broker_decoder=decoder, broker_codec=codec, broker_parser=parser, - broker_exception_handler=( - to_async(exception_handler) if exception_handler is not None else None - ), + broker_exception_handler=ensure_exception_handler(exception_handler), broker_middlewares=middlewares, logger=make_kafka_logger_state( logger=logger, diff --git a/faststream/kafka/subscriber/factory.py b/faststream/kafka/subscriber/factory.py index 9463c59532a..66159ef976d 100644 --- a/faststream/kafka/subscriber/factory.py +++ b/faststream/kafka/subscriber/factory.py @@ -4,7 +4,7 @@ from faststream._internal.constants import EMPTY from faststream._internal.endpoint.subscriber.call_item import CallsCollection -from faststream._internal.utils import to_async +from faststream._internal.utils import ensure_exception_handler from faststream.exceptions import SetupError from faststream.kafka._compat import validate_client_rack from faststream.kafka.schemas import Topic @@ -74,9 +74,7 @@ def create_subscriber( listener=listener, pattern=pattern, no_reply=no_reply, - exception_handler=( - to_async(exception_handler) if exception_handler is not None else None - ), + exception_handler=ensure_exception_handler(exception_handler), _outer_config=config, _ack_policy=ack_policy, ) diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py index 8e882448bbb..d52524cfaf6 100644 --- a/tests/brokers/base/exception_handlers.py +++ b/tests/brokers/base/exception_handlers.py @@ -4,19 +4,71 @@ import pytest from faststream import Context, FastStream, TestApp +from faststream.exceptions import StopApplication, StopConsume from tests.brokers.base.basic import BaseTestcaseConfig @pytest.mark.asyncio() class ExceptionHandlersTestcase(BaseTestcaseConfig[Any]): - async def get_consume_message( + async def test_exception_handlers_are_not_called_for_successful_message( self, - broker: Any, - body: Any, queue: str, - ) -> Any: - """Build a raw message; Kafka overrides this with a ConsumerRecord.""" - raise NotImplementedError + mock: MagicMock, + mock2: MagicMock, + ) -> None: + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + mock2(msg) + + app = FastStream(broker, exception_handler=exception_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish("hello", queue) + + mock2.assert_called_once_with("hello") + mock.assert_not_called() + + @pytest.mark.parametrize("error_type", (StopConsume, StopApplication)) + async def test_exception_handlers_do_not_suppress_stop_commands( + self, + queue: str, + mock: MagicMock, + error_type: type[BaseException], + ) -> None: + error = error_type() + + async def exception_handler(exc: BaseException) -> bool: + mock(exc) + return True + + broker = self.get_broker(exception_handler=exception_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=exception_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises(error_type) as exc_info: + await br.publish(None, queue) + + assert exc_info.value is error + mock.assert_not_called() async def test_subscriber_exception_handler_true_suppresses_error( self, @@ -41,8 +93,7 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) @@ -75,8 +126,7 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error, expected_service) @@ -108,8 +158,7 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) mock2.assert_not_called() @@ -141,8 +190,7 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) assert mock.call_args_list == [ call("subscriber", error), @@ -169,8 +217,7 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) @@ -203,8 +250,7 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error, expected_service) @@ -231,12 +277,11 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) with pytest.raises( RuntimeError, match="message processing failed", ) as exc_info: - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) assert exc_info.value is error @@ -268,12 +313,11 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) with pytest.raises( RuntimeError, match="message processing failed", ) as exc_info: - await subscriber.consume(message) + await br.publish(None, queue) assert mock.call_args_list == [ call("subscriber", error), @@ -301,12 +345,11 @@ async def handler(msg: Any) -> None: raise error async with self.patch_broker(broker) as br: - message = await self.get_consume_message(br, "hello", queue) with pytest.raises( RuntimeError, match="message processing failed", ) as exc_info: - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) assert exc_info.value is error @@ -351,13 +394,11 @@ async def handler2(msg: Any) -> None: self.patch_broker(broker2, connect_only=True) as br2, TestApp(app), ): - message1 = await self.get_consume_message(br1, "hello", queue) - await subscriber1.consume(message1) + await br1.publish(None, queue) mock2.assert_not_called() - message2 = await self.get_consume_message(br2, "hello", queue + "1") - await subscriber2.consume(message2) + await br2.publish(None, queue + "1") mock.assert_called_once_with(error1) mock2.assert_called_once_with(error2) @@ -395,8 +436,7 @@ async def handler(msg: Any) -> None: ) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) @@ -422,12 +462,11 @@ async def handler(msg: Any) -> None: app = FastStream(broker, exception_handler=exception_handler) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) with pytest.raises( RuntimeError, match="message processing failed", ) as exc_info: - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error) assert exc_info.value is error @@ -465,8 +504,7 @@ async def handler(msg: Any) -> None: app = FastStream(broker, exception_handler=app_handler) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) assert mock.call_args_list == [ call("subscriber", error), @@ -510,8 +548,7 @@ async def handler(msg: Any) -> None: app = FastStream(broker, exception_handler=app_handler) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) expected_calls = [call("subscriber", error)] if handled_by == "broker": @@ -553,12 +590,11 @@ async def handler(msg: Any) -> None: app = FastStream(broker, exception_handler=app_handler) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) with pytest.raises( RuntimeError, match="message processing failed", ) as exc_info: - await subscriber.consume(message) + await br.publish(None, queue) assert mock.call_args_list == [ call("subscriber", error), @@ -594,8 +630,7 @@ async def handler(msg: Any) -> None: app.context.set_global("error_service", expected_service) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error, expected_service) @@ -639,17 +674,15 @@ async def handler2(msg: Any) -> None: TestApp(app1), TestApp(app2), ): - message1 = await self.get_consume_message(br1, "hello", queue) - await subscriber1.consume(message1) + await br1.publish(None, queue) mock2.assert_not_called() - message2 = await self.get_consume_message(br2, "hello", queue + "1") - await subscriber2.consume(message2) + await br2.publish(None, queue + "1") mock.assert_called_once_with(error1) mock2.assert_called_once_with(error2) - async def test_app_without_exception_handlers_preserves_error_suppression( + async def test_app_without_exception_handlers_propagates_error( self, queue: str, mock: MagicMock, @@ -667,10 +700,14 @@ async def handler(msg: Any) -> None: app = FastStream(broker) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + with pytest.raises( + RuntimeError, + match="message processing failed", + ) as exc_info: + await br.publish("hello", queue) mock.assert_called_once_with("hello") + assert exc_info.value is error @pytest.mark.parametrize("handler_level", ("subscriber", "broker")) async def test_exception_handler_receives_dependency_from_app_context( @@ -712,7 +749,49 @@ async def handler(msg: Any) -> None: app.context.set_global("error_service", expected_service) async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): - message = await self.get_consume_message(br, "hello", queue) - await subscriber.consume(message) + await br.publish(None, queue) mock.assert_called_once_with(error, expected_service) + + async def test_app_exception_handler_receives_each_broker_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error1 = RuntimeError("first broker processing failed") + error2 = RuntimeError("second broker processing failed") + + async def exception_handler( + exc: BaseException, + current_broker: Any = Context("broker"), + ) -> bool: + mock(exc, current_broker) + return True + + broker1 = self.get_broker(apply_types=True) + broker2 = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params(queue) + args2, kwargs2 = self.get_subscriber_params(queue + "1") + + @broker1.subscriber(*args, **kwargs) + async def handler1(msg: Any) -> None: + raise error1 + + @broker2.subscriber(*args2, **kwargs2) + async def handler2(msg: Any) -> None: + raise error2 + + app = FastStream(broker1, broker2, exception_handler=exception_handler) + + async with ( + self.patch_broker(broker1, connect_only=True) as br1, + self.patch_broker(broker2, connect_only=True) as br2, + TestApp(app), + ): + await br1.publish(None, queue) + await br2.publish(None, queue + "1") + + assert mock.call_args_list == [ + call(error1, broker1), + call(error2, broker2), + ] diff --git a/tests/brokers/kafka/test_exception_handlers.py b/tests/brokers/kafka/test_exception_handlers.py index 3571a099f03..fe7082fa35c 100644 --- a/tests/brokers/kafka/test_exception_handlers.py +++ b/tests/brokers/kafka/test_exception_handlers.py @@ -1,32 +1,9 @@ -from typing import TYPE_CHECKING, Any - import pytest -from typing_extensions import override -from faststream.kafka.testing import build_message from tests.brokers.base.exception_handlers import ExceptionHandlersTestcase from .basic import KafkaMemoryTestcaseConfig -if TYPE_CHECKING: - from aiokafka import ConsumerRecord - - from faststream.kafka import KafkaBroker - @pytest.mark.kafka() -class TestExceptionHandlers(KafkaMemoryTestcaseConfig, ExceptionHandlersTestcase): - @override - async def get_consume_message( - self, - broker: "KafkaBroker", - body: Any, - queue: str, - ) -> "ConsumerRecord": - return await build_message( - message=body, - topic=queue, - serializer=broker.config.fd_config._serializer, - codec=broker.config.broker_codec, - id_generator=broker.config.id_generator, - ) +class TestExceptionHandlers(KafkaMemoryTestcaseConfig, ExceptionHandlersTestcase): ... diff --git a/tests/docs/getting_started/exception_handlers/__init__.py b/tests/docs/getting_started/exception_handlers/__init__.py new file mode 100644 index 00000000000..8b137891791 --- /dev/null +++ b/tests/docs/getting_started/exception_handlers/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/docs/getting_started/exception_handlers/test_handlers.py b/tests/docs/getting_started/exception_handlers/test_handlers.py new file mode 100644 index 00000000000..53f3e83b962 --- /dev/null +++ b/tests/docs/getting_started/exception_handlers/test_handlers.py @@ -0,0 +1,32 @@ +import pytest + +from faststream import TestApp +from tests.marks import require_aiokafka + + +@pytest.mark.kafka() +@pytest.mark.asyncio() +@require_aiokafka +@pytest.mark.parametrize("message", ("subscriber", "broker")) +async def test_kafka_example_suppresses_handled_errors(message: str) -> None: + from docs.docs_src.getting_started.exception_handlers.kafka import app, broker, handle + from faststream.kafka import TestKafkaBroker + + async with TestKafkaBroker(broker, connect_only=True) as br, TestApp(app): + await br.publish(message, "exception-handlers") + + handle.mock.assert_called_once_with(message) + + +@pytest.mark.kafka() +@pytest.mark.asyncio() +@require_aiokafka +async def test_kafka_example_propagates_error_after_app_declines() -> None: + from docs.docs_src.getting_started.exception_handlers.kafka import app, broker, handle + from faststream.kafka import TestKafkaBroker + + async with TestKafkaBroker(broker, connect_only=True) as br, TestApp(app): + with pytest.raises(TypeError, match="application error"): + await br.publish("application", "exception-handlers") + + handle.mock.assert_called_once_with("application") From 8664e378c01cdec0eb4ced81c3a75b6f08019bc8 Mon Sep 17 00:00:00 2001 From: DMINQ <213875675+DMINQ@users.noreply.github.com> Date: Sun, 25 Oct 2026 01:11:05 +0300 Subject: [PATCH 7/7] fix: correct exception handler context and acknowledgement order --- faststream/_internal/broker/broker.py | 3 + faststream/_internal/broker/registrator.py | 3 + .../_internal/endpoint/subscriber/usecase.py | 56 ++-- faststream/app.py | 5 + tests/brokers/base/exception_handlers.py | 296 ++++++++++++++++-- 5 files changed, 319 insertions(+), 44 deletions(-) diff --git a/faststream/_internal/broker/broker.py b/faststream/_internal/broker/broker.py index cceef1b080d..4b4ece24450 100644 --- a/faststream/_internal/broker/broker.py +++ b/faststream/_internal/broker/broker.py @@ -95,6 +95,9 @@ def _update_fd_config(self, config: "FastDependsConfig") -> None: """Private method to change broker config state by outer application.""" self.config.fd_config = config | self.config.fd_config + for subscriber in self.subscribers: + subscriber._setup_exception_handlers() + async def start(self) -> None: # TODO: filter by already running handlers after TestClient refactor for sub in self.subscribers: diff --git a/faststream/_internal/broker/registrator.py b/faststream/_internal/broker/registrator.py index 7704be0da51..53532cf3a06 100644 --- a/faststream/_internal/broker/registrator.py +++ b/faststream/_internal/broker/registrator.py @@ -114,6 +114,9 @@ def include_router( router.config.add_config(self.config) self.routers.append(router) + for subscriber in router.subscribers: + subscriber._setup_exception_handlers() + @property def parent(self) -> "Registrator[MsgType, Any] | None": return self.__parent diff --git a/faststream/_internal/endpoint/subscriber/usecase.py b/faststream/_internal/endpoint/subscriber/usecase.py index 2ba8330625c..e0f7895ce75 100644 --- a/faststream/_internal/endpoint/subscriber/usecase.py +++ b/faststream/_internal/endpoint/subscriber/usecase.py @@ -27,9 +27,9 @@ P_HandlerParams, T_HandlerReturn, ) -from faststream._internal.utils.functions import to_async +from faststream._internal.utils import apply_types, to_async from faststream.exceptions import StopConsume, SubscriberNotFound -from faststream.middlewares import AcknowledgementMiddleware +from faststream.middlewares import AcknowledgementMiddleware, ExceptionMiddleware from faststream.middlewares.logging import CriticalLogMiddleware from faststream.response import ensure_response @@ -77,6 +77,8 @@ class SubscriberUsecase(Endpoint, Generic[MsgType]): "_call_decorators", "_call_options", "_decoder", + "_exception_handlers", + "_exception_middleware", "_no_reply", "_parser", "ack_policy", @@ -86,6 +88,8 @@ class SubscriberUsecase(Endpoint, Generic[MsgType]): "specification", ) + _exception_handlers: tuple["AsyncExceptionHandler", ...] + lock: "AbstractContextManager[Any]" graceful_timeout: float | None @@ -121,6 +125,11 @@ def __init__( self.running = False self.lock = nullcontext() + self._setup_exception_handlers() + self._exception_middleware = ExceptionMiddleware( + handlers={Exception: self._handle_exception}, + ) + @property def _broker_middlewares(self) -> Sequence["BrokerMiddleware[MsgType]"]: return self._outer_config.broker_middlewares @@ -349,26 +358,29 @@ def _app_exception_handler(self) -> "AsyncExceptionHandler | None": ) return app_handler - async def _handle_exception( - self, - _exc_type: type[BaseException] | None, - exc: BaseException | None, - _exc_tb: TracebackType | None, - ) -> bool: - if not isinstance(exc, Exception) or isinstance(exc, (StopConsume, SystemExit)): - return False - - context = self._outer_config.context + def _setup_exception_handlers(self) -> None: + config = self._outer_config.fd_config + self._exception_handlers = tuple( + apply_types( + handler, + serializer_cls=config._serializer, + dependency_provider=config.provider, + context__=config.context, + ) + for handler in ( + self._exception_handler, + self._outer_config.broker_exception_handler, + self._app_exception_handler, + ) + if handler is not None + ) - for handler in ( - self._exception_handler, - self._outer_config.broker_exception_handler, - self._app_exception_handler, - ): - if handler is not None and await handler(exc, context__=context): - return True + async def _handle_exception(self, exc: Exception) -> None: + for handler in self._exception_handlers: + if await handler(exc): + return - return False + raise exc async def consume(self, msg: MsgType) -> Any: """Consume a message asynchronously.""" @@ -411,7 +423,6 @@ async def process_message(self, msg: MsgType) -> "Response": # reserve place for context scope under the middlewares __aexit__ # because middlewares should be exited before context scope release message_scope = stack.enter_context(ExitStack()) - stack.push_async_exit(self._handle_exception) # enter all middlewares middlewares: list[BaseMiddleware] = [] @@ -480,10 +491,12 @@ async def process_message(self, msg: MsgType) -> "Response": def __build__middlewares_stack(self) -> tuple["BrokerMiddleware[MsgType]", ...]: logger_state = self._outer_config.logger + broker_middlewares: tuple[BrokerMiddleware[MsgType], ...] if self.__auto_ack_disabled: broker_middlewares = ( CriticalLogMiddleware(logger_state), + self._exception_middleware, *self._broker_middlewares, ) @@ -495,6 +508,7 @@ def __build__middlewares_stack(self) -> tuple["BrokerMiddleware[MsgType]", ...]: fallback_parser=self._parser, ), CriticalLogMiddleware(logger_state), + self._exception_middleware, *self._broker_middlewares, ) diff --git a/faststream/app.py b/faststream/app.py index 86c4b109329..4f1d321fe00 100644 --- a/faststream/app.py +++ b/faststream/app.py @@ -85,6 +85,11 @@ def __init__( self._should_exit = False + # super().__init__ registers brokers before the application handler exists. + for broker in self.brokers: + for subscriber in broker.subscribers: + subscriber._setup_exception_handlers() + async def run( self, log_level: int = logging.INFO, diff --git a/tests/brokers/base/exception_handlers.py b/tests/brokers/base/exception_handlers.py index d52524cfaf6..0de10d3f501 100644 --- a/tests/brokers/base/exception_handlers.py +++ b/tests/brokers/base/exception_handlers.py @@ -5,6 +5,8 @@ from faststream import Context, FastStream, TestApp from faststream.exceptions import StopApplication, StopConsume +from faststream.message import AckStatus, StreamMessage +from faststream.middlewares import AckPolicy from tests.brokers.base.basic import BaseTestcaseConfig @@ -512,19 +514,17 @@ async def handler(msg: Any) -> None: call("app", error), ] - @pytest.mark.parametrize("handled_by", ("subscriber", "broker")) - async def test_app_handler_is_not_called_after_error_is_handled( + async def test_app_handler_is_not_called_after_subscriber_handles_error( self, queue: str, mock: MagicMock, mock2: MagicMock, - handled_by: str, ) -> None: error = RuntimeError("message processing failed") async def local_handler(exc: BaseException) -> bool: mock("subscriber", exc) - return handled_by == "subscriber" + return True async def broker_handler(exc: BaseException) -> bool: mock("broker", exc) @@ -550,11 +550,49 @@ async def handler(msg: Any) -> None: async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): await br.publish(None, queue) - expected_calls = [call("subscriber", error)] - if handled_by == "broker": - expected_calls.append(call("broker", error)) + assert mock.call_args_list == [call("subscriber", error)] + mock2.assert_not_called() + + async def test_app_handler_is_not_called_after_broker_handles_error( + self, + queue: str, + mock: MagicMock, + mock2: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + + async def local_handler(exc: BaseException) -> bool: + mock("subscriber", exc) + return False - assert mock.call_args_list == expected_calls + async def broker_handler(exc: BaseException) -> bool: + mock("broker", exc) + return True + + async def app_handler(exc: BaseException) -> bool: + mock2(exc) + return True + + broker = self.get_broker(exception_handler=broker_handler) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker, exception_handler=app_handler) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error), + call("broker", error), + ] mock2.assert_not_called() async def test_error_propagates_when_all_three_handlers_decline( @@ -709,12 +747,10 @@ async def handler(msg: Any) -> None: mock.assert_called_once_with("hello") assert exc_info.value is error - @pytest.mark.parametrize("handler_level", ("subscriber", "broker")) - async def test_exception_handler_receives_dependency_from_app_context( + async def test_subscriber_exception_handler_receives_dependency_from_app_context( self, queue: str, mock: MagicMock, - handler_level: str, ) -> None: error = RuntimeError("message processing failed") expected_service = object() @@ -726,18 +762,46 @@ async def exception_handler( mock(exc, service) return True - if handler_level == "broker": - broker = self.get_broker( - apply_types=True, - exception_handler=exception_handler, - ) - args, kwargs = self.get_subscriber_params(queue) - else: - broker = self.get_broker(apply_types=True) - args, kwargs = self.get_subscriber_params( - queue, - exception_handler=exception_handler, - ) + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=exception_handler, + ) + + subscriber = broker.subscriber(*args, **kwargs) + + @subscriber + async def handler(msg: Any) -> None: + raise error + + app = FastStream(broker) + app.context.set_global("error_service", expected_service) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + mock.assert_called_once_with(error, expected_service) + + async def test_broker_exception_handler_receives_dependency_from_app_context( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def exception_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock(exc, service) + return True + + broker = self.get_broker( + apply_types=True, + exception_handler=exception_handler, + ) + args, kwargs = self.get_subscriber_params(queue) subscriber = broker.subscriber(*args, **kwargs) @@ -795,3 +859,189 @@ async def handler2(msg: Any) -> None: call(error1, broker1), call(error2, broker2), ] + + async def test_exception_handlers_receive_context_after_broker_is_added( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def local_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("subscriber", exc, service) + return False + + async def broker_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("broker", exc, service) + return False + + async def app_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("app", exc, service) + return True + + broker = self.get_broker( + apply_types=True, + exception_handler=broker_handler, + ) + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + raise error + + app = FastStream(exception_handler=app_handler) + app.context.set_global("error_service", expected_service) + app.add_broker(broker) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error, expected_service), + call("broker", error, expected_service), + call("app", error, expected_service), + ] + + async def test_exception_handlers_receive_context_after_router_is_included( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + expected_service = object() + + async def local_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("subscriber", exc, service) + return False + + async def broker_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("broker", exc, service) + return False + + async def app_handler( + exc: BaseException, + service: object = Context("error_service"), + ) -> bool: + mock("app", exc, service) + return True + + broker = self.get_broker(apply_types=True, exception_handler=broker_handler) + app = FastStream(broker, exception_handler=app_handler) + app.context.set_global("error_service", expected_service) + router = self.get_router() + args, kwargs = self.get_subscriber_params( + queue, + exception_handler=local_handler, + ) + + @router.subscriber(*args, **kwargs) + async def handler(msg: Any) -> None: + raise error + + broker.include_router(router) + + async with self.patch_broker(broker, connect_only=True) as br, TestApp(app): + await br.publish(None, queue) + + assert mock.call_args_list == [ + call("subscriber", error, expected_service), + call("broker", error, expected_service), + call("app", error, expected_service), + ] + + async def test_handled_error_is_suppressed_before_acknowledgement( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + messages: list[StreamMessage[Any]] = [] + + async def exception_handler( + exc: BaseException, + message: StreamMessage[Any] = Context("message"), + ) -> bool: + mock(exc, message.committed) + return True + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + ack_policy=AckPolicy.NACK_ON_ERROR, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler( + msg: Any, + message: StreamMessage[Any] = Context("message"), + ) -> None: + messages.append(message) + raise error + + async with self.patch_broker(broker) as br: + await br.publish(None, queue) + + assert (mock.call_args_list, [message.committed for message in messages]) == ( + [call(error, None)], + [AckStatus.ACKED], + ) + + async def test_unhandled_error_is_propagated_after_negative_acknowledgement( + self, + queue: str, + mock: MagicMock, + ) -> None: + error = RuntimeError("message processing failed") + messages: list[StreamMessage[Any]] = [] + + async def exception_handler( + exc: BaseException, + message: StreamMessage[Any] = Context("message"), + ) -> bool: + mock(exc, message.committed) + return False + + broker = self.get_broker(apply_types=True) + args, kwargs = self.get_subscriber_params( + queue, + ack_policy=AckPolicy.NACK_ON_ERROR, + exception_handler=exception_handler, + ) + + @broker.subscriber(*args, **kwargs) + async def handler( + msg: Any, + message: StreamMessage[Any] = Context("message"), + ) -> None: + messages.append(message) + raise error + + async with self.patch_broker(broker) as br: + with pytest.raises(RuntimeError) as exc_info: + await br.publish(None, queue) + + assert ( + mock.call_args_list, + [message.committed for message in messages], + exc_info.value, + ) == ([call(error, None)], [AckStatus.NACKED], error)