diff --git a/modern_di_taskiq/main.py b/modern_di_taskiq/main.py index 83bc4b1..d9abeb4 100644 --- a/modern_di_taskiq/main.py +++ b/modern_di_taskiq/main.py @@ -83,7 +83,11 @@ async def build_di_container( async with fetch_di_container(context.broker).build_child_container( scope=match.scope, context=match.context ) as container: - yield container + try: + yield container + except Exception: # noqa: BLE001 + # taskiq throws the task's own error in here and logs it again if it comes back out. + return @dataclasses.dataclass(slots=True, frozen=True) diff --git a/tests/test_taskiq_di.py b/tests/test_taskiq_di.py index 5143bc6..4b5a532 100644 --- a/tests/test_taskiq_di.py +++ b/tests/test_taskiq_di.py @@ -1,6 +1,8 @@ +import logging import re import typing +import pytest from modern_di import Container, Group, Scope, providers from taskiq import InMemoryBroker @@ -119,6 +121,25 @@ async def boom(_res: typing.Annotated[SimpleCreator, FromDI(Boom.resource)]) -> assert teardowns == ["closed"] # per-task child was still closed (finalizer ran) on the error path +async def test_task_error_is_not_logged_again_as_a_teardown_error( + broker: InMemoryBroker, caplog: pytest.LogCaptureFixture +) -> None: + @broker.task(task_name="fails") + async def fails(_dep: typing.Annotated[SimpleCreator, FromDI(SimpleCreator)]) -> None: + msg = "kaboom" + raise ValueError(msg) + + await broker.startup() + try: + with caplog.at_level(logging.WARNING): + result = await (await fails.kiq()).wait_result() # ty: ignore[no-matching-overload] + finally: + await broker.shutdown() + + assert result.is_err is True + assert [r.getMessage() for r in caplog.records if r.name == "taskiq.dependencies.ctx"] == [] + + async def test_from_di_without_setup_di_raises_clear_runtime_error() -> None: broker = InMemoryBroker()