docs: add examples with ProgressTracker

Author: danfimov

docs/examples/state/progress_tracker.py

@@ -0,0 +1,11 @@
+import asyncio
+
+from taskiq_dependencies import Depends
+
+from taskiq.depends.progress_tracker import ProgressTracker, TaskState
+
+
+async def my_task(progress: ProgressTracker[str] = Depends()) -> None:
+    for i in range(10):
+        await asyncio.sleep(1)
+        await progress.set_progress(TaskState.STARTED, meta=f"{(i + 1) * 10}%")

docs/examples/state/progress_tracker_annot.py

@@ -0,0 +1,12 @@
+import asyncio
+from typing import Annotated
+
+from taskiq_dependencies import Depends
+
+from taskiq.depends.progress_tracker import ProgressTracker, TaskState
+
+
+async def my_task(progress: Annotated[ProgressTracker[str], Depends()]) -> None:
+    for i in range(10):
+        await asyncio.sleep(1)
+        await progress.set_progress(TaskState.STARTED, meta=f"{(i + 1) * 10}%")

docs/guide/state-and-deps.md

@@ -319,6 +319,48 @@ taskiq worker my_file:broker --no-propagate-errors
 
 In this case, no exception will ever going to be propagated to any dependency.
 
+## Progress tracking
+
+Sometimes a task runs for a long time and you want to report how far it has progressed, so other parts of your system
+(e.g. a web handler polling for status) can display it.
+
+Taskiq provides a `ProgressTracker` dependency for this. It's not a method on `Context`, it's a dependency, just like
+anything else built with `taskiq-dependencies`. It grabs the current `task_id` from the context for you and stores
+progress using your broker's result backend.
+
+::: tabs
+
+@tab Annotated 3.10+
+
+@[code python](../examples/state/progress_tracker_annot.py)
+
+@tab default values
+
+@[code python](../examples/state/progress_tracker.py)
+
+:::
+
+You can read the progress back from anywhere that has access to the `AsyncTaskiqTask` returned by `kiq`:
+
+```python
+task = await my_task.kiq()
+
+progress = await task.get_progress()
+if progress is not None:
+    print(progress.state, progress.meta)
+```
+
+`state` can be one of the `TaskState` enum values (`STARTED`, `SUCCESS`, `FAILURE`, `RETRY`) or any custom string.
+`meta` is generic (`ProgressTracker[MetaType]`) and can be any value your result backend can serialize, such as a plain
+string, a `dict`, or a pydantic model. If you call `set_progress` without `meta`, the previously stored `meta` value is
+preserved, which is handy when you only want to update `state`.
+
+::: warning important note
+
+`set_progress`/`get_progress` are no-ops by default on `AsyncResultBackend`. Make sure the result backend you use
+actually implements progress storage (`InMemoryBroker`'s built-in backend does), before relying on this in production.
+
+:::
 
 ## Generics
 

taskiq/cli/scheduler/run.py

@@ -313,7 +313,7 @@ def _is_schedule_ready_to_send(
 
         return is_ready_to_send
 
-    async def run(
+    async def run(  # noqa: C901
         self,
         *,
         update_interval: timedelta | None = None,

tests/cli/scheduler/test_send_with_timeout.py

@@ -31,7 +31,9 @@ def __init__(self, hang_seconds: float = 60.0) -> None:
         self.on_ready_calls = 0
 
     async def on_ready(  # type: ignore[override]
-        self, source: ScheduleSource, task: ScheduledTask
+        self,
+        source: ScheduleSource,
+        task: ScheduledTask,
     ) -> None:
         self.on_ready_calls += 1
         await asyncio.sleep(self.hang_seconds)
@@ -98,7 +100,9 @@ def __init__(self) -> None:
             self.calls = 0
 
         async def on_ready(  # type: ignore[override]
-            self, source: ScheduleSource, task: ScheduledTask
+            self,
+            source: ScheduleSource,
+            task: ScheduledTask,
         ) -> None:
             self.calls += 1
 
@@ -131,7 +135,9 @@ def __init__(self) -> None:
             super().__init__(broker=InMemoryBroker(), sources=[_StubSource()])
 
         async def on_ready(  # type: ignore[override]
-            self, source: ScheduleSource, task: ScheduledTask
+            self,
+            source: ScheduleSource,
+            task: ScheduledTask,
         ) -> None:
             raise RuntimeError("boom")
 
@@ -153,7 +159,9 @@ def __init__(self) -> None:
             super().__init__(broker=InMemoryBroker(), sources=[_StubSource()])
 
         async def on_ready(  # type: ignore[override]
-            self, source: ScheduleSource, task: ScheduledTask
+            self,
+            source: ScheduleSource,
+            task: ScheduledTask,
         ) -> None:
             try:
                 await asyncio.sleep(30.0)
@@ -184,7 +192,9 @@ def __init__(self) -> None:
             self.calls: list[Any] = []
 
         async def on_ready(  # type: ignore[override]
-            self, source: ScheduleSource, task: ScheduledTask
+            self,
+            source: ScheduleSource,
+            task: ScheduledTask,
         ) -> None:
             self.calls.append((source, task))