Added exception catches from dependencies. (#153)

Author: s3rius

.flake8

@@ -95,17 +95,6 @@ ignore =
   ; Found too many public instance attributes
   WPS230,
 
-  ; all init files
-  __init__.py:
-  ; ignore not used imports
-  F401,
-  ; ignore import with wildcard
-  F403,
-  ; Found wrong metadata variable
-  WPS410,
-  ; Found commented out cod
-  E800,
-
 per-file-ignores =
   ; all tests
   test_*.py,tests.py,tests_*.py,*/tests/*:
@@ -120,6 +109,21 @@ per-file-ignores =
   ; Found complex default value
   WPS404,
 
+  ; all init files
+  __init__.py:
+  ; ignore not used imports
+  F401,
+  ; ignore import with wildcard
+  F403,
+  ; Found wrong metadata variable
+  WPS410,
+  ; Found commented out cod
+  E800,
+
+  taskiq/serialization.py:
+  ; Found commented out code
+  E800,
+
 exclude =
   ./.git,
   ./venv,

taskiq/abc/broker.py

@@ -86,6 +86,7 @@ def __init__(
                 "Setting result backend with constructor is deprecated. "
                 "Please use `with_result_backend` instead.",
                 TaskiqDeprecationWarning,
+                stacklevel=2,
             )
         if task_id_generator is None:
             task_id_generator = default_id_generator
@@ -94,6 +95,7 @@ def __init__(
                 "Setting id generator with constructor is deprecated. "
                 "Please use `with_id_generator` instead.",
                 TaskiqDeprecationWarning,
+                stacklevel=2,
             )
         self.middlewares: "List[TaskiqMiddleware]" = []
         self.result_backend = result_backend

taskiq/receiver/receiver.py

@@ -206,14 +206,19 @@ async def run_task(  # noqa: C901, WPS210
             )
             dep_ctx = dependency_graph.async_ctx(broker_ctx)
             # Resolve all function's dependencies.
-            kwargs = await dep_ctx.resolve_kwargs()
-
-        # We udpate kwargs with kwargs from network.
-        kwargs.update(message.kwargs)
 
         # Start a timer.
         start_time = time()
+
         try:
+            # We put kwargs resolving here,
+            # to be able to catch any exception (for example ),
+            # that happen while resolving dependencies.
+            if dep_ctx:
+                kwargs = await dep_ctx.resolve_kwargs()
+            # We udpate kwargs with kwargs from network.
+            kwargs.update(message.kwargs)
+
             # If the function is a coroutine, we await it.
             if asyncio.iscoroutinefunction(target):
                 returned = await target(*message.args, **kwargs)

tests/conftest.py

@@ -1,5 +1,9 @@
+from typing import Generator
+
 import pytest
 
+from taskiq.abc.broker import AsyncBroker
+
 
 @pytest.fixture(scope="session")
 def anyio_backend() -> str:
@@ -10,3 +14,17 @@ def anyio_backend() -> str:
     :return: backend name.
     """
     return "asyncio"
+
+
+@pytest.fixture(autouse=True)
+def reset_broker() -> Generator[None, None, None]:
+    """
+    Restore async broker.
+
+    This fixtures sets some global
+    broker variables to default state.
+    """
+    yield
+    AsyncBroker.available_tasks = {}
+    AsyncBroker.is_worker_process = False
+    AsyncBroker.is_scheduler_process = False

tests/test_requeue.py

@@ -20,3 +20,25 @@ async def task(context: Context = TaskiqDepends()) -> None:
     await kicked.wait_result()
 
     assert runs_count == 2
+
+
+@pytest.mark.anyio
+async def test_requeue_from_dependency() -> None:
+    broker = InMemoryBroker()
+
+    runs_count = 0
+
+    async def dep_func(context: Context = TaskiqDepends()) -> None:
+        nonlocal runs_count
+        runs_count += 1
+        if runs_count < 2:
+            await context.requeue()
+
+    @broker.task
+    async def task(_: None = TaskiqDepends(dep_func)) -> None:
+        return None
+
+    kicked = await task.kiq()
+    await kicked.wait_result()
+
+    assert runs_count == 2