Merge branch 'release/0.7.2'

Author: s3rius

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskiq"
-version = "0.7.1"
+version = "0.7.2"
 description = "Distributed task queue with full async support"
 authors = ["Pavel Kirilin <win10@list.ru>"]
 maintainers = ["Pavel Kirilin <win10@list.ru>"]

taskiq/context.py

@@ -1,4 +1,3 @@
-from copy import copy
 from typing import TYPE_CHECKING
 
 from taskiq.abc.broker import AsyncBroker
@@ -28,10 +27,9 @@ async def requeue(self) -> None:
 
         :raises NoResultError: to not store result for current task.
         """
-        message = copy(self.message)
-        requeue_count = int(message.labels.get("X-Taskiq-requeue", 0))
+        requeue_count = int(self.message.labels.get("X-Taskiq-requeue", 0))
         requeue_count += 1
-        message.labels["X-Taskiq-requeue"] = str(requeue_count)
+        self.message.labels["X-Taskiq-requeue"] = str(requeue_count)
         await self.broker.kick(self.broker.formatter.dumps(self.message))
         raise NoResultError()
 

taskiq/receiver/receiver.py

@@ -265,6 +265,7 @@ async def run_task(  # noqa: C901, WPS210
             return_value=returned,
             execution_time=round(execution_time, 2),
             error=found_exception,
+            labels=message.labels,
         )
         # If exception is found we execute middlewares.
         if found_exception is not None:

taskiq/result.py

@@ -3,7 +3,7 @@
 from functools import partial
 from typing import Any, Callable, Dict, Generic, Optional, TypeVar
 
-from pydantic import validator
+from pydantic import Field, validator
 from pydantic.generics import GenericModel
 from typing_extensions import Self
 
@@ -33,6 +33,7 @@ class TaskiqResult(GenericModel, Generic[_ReturnType]):
     log: Optional[str] = None
     return_value: _ReturnType
     execution_time: float
+    labels: Dict[str, str] = Field(default_factory=dict)
 
     error: Optional[BaseException] = None
 

tests/test_requeue.py

@@ -18,6 +18,10 @@ async def task(context: Context = TaskiqDepends()) -> None:
 
     kicked = await task.kiq()
     await kicked.wait_result()
+    assert (
+        broker.custom_dependency_context[Context].message.labels["X-Taskiq-requeue"]
+        == "1"
+    )
 
     assert runs_count == 2
 
@@ -40,5 +44,9 @@ async def task(_: None = TaskiqDepends(dep_func)) -> None:
 
     kicked = await task.kiq()
     await kicked.wait_result()
+    assert (
+        broker.custom_dependency_context[Context].message.labels["X-Taskiq-requeue"]
+        == "1"
+    )
 
     assert runs_count == 2