Merge branch 'master' into chore-replace-pre-commit-with-prek

Author: danfimov

docs/.vuepress/config.ts

@@ -33,6 +33,7 @@ export default defineUserConfig({
     sidebar: "structure",
 
     pure: true,
+    focus: false,
 
     markdown: {
       tabs: true,

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

@@ -4,7 +4,7 @@
 from datetime import datetime, timedelta, timezone
 from logging import basicConfig, getLogger
 from typing import Any, TypeAlias
-from zoneinfo import ZoneInfo
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 import pycron
 
@@ -99,7 +99,13 @@ def is_cron_task_now(
     # If timezone was specified as string we convert it timezone
     # offset and then apply.
     elif offset and isinstance(offset, str):
-        now = now.astimezone(ZoneInfo(offset))
+        try:
+            now = now.astimezone(ZoneInfo(offset))
+        # ZoneInfoNotFoundError for unknown keys, ModuleNotFoundError
+        # for systems without timezone data available (e.g. missing
+        # tzdata on Windows).
+        except (ZoneInfoNotFoundError, ModuleNotFoundError) as e:
+            raise CronValueError(e) from e
 
     try:
         return pycron.is_now(cron_value, now)

tests/abc/test_broker.py

@@ -82,7 +82,6 @@ async def test_task() -> None: ...
     assert test_task.labels == old_labels
 
 
-@pytest.mark.anyio
 @pytest.mark.parametrize(
     ("is_worker_process", "startup", "shutdown"),
     [
@@ -120,7 +119,6 @@ async def track_shutdown(state: TaskiqState) -> None:
     assert shutdown_called is True
 
 
-@pytest.mark.anyio
 @pytest.mark.parametrize(
     ("is_worker_process", "startup", "shutdown"),
     [

tests/cli/scheduler/test_is_cron_task_now.py

@@ -75,3 +75,16 @@ def test_is_cron_task_now(
 def test_is_cron_task_now_invalid_cron() -> None:
     with pytest.raises(CronValueError):
         is_cron_task_now("invalid cron", datetime.now())
+
+
+@pytest.mark.parametrize(
+    "offset",
+    ["UTC+3", "Europa/Madrid"],
+)
+def test_is_cron_task_now_invalid_offset_string(offset: str) -> None:
+    with pytest.raises(CronValueError):
+        is_cron_task_now(
+            cron_value="* * * * *",
+            now=datetime.now(timezone.utc),
+            offset=offset,
+        )

tests/cli/scheduler/test_send_with_timeout.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 import pytest
+from typing_extensions import override
 
 from taskiq.abc.schedule_source import ScheduleSource
 from taskiq.brokers.inmemory_broker import InMemoryBroker
@@ -30,7 +31,8 @@ def __init__(self, hang_seconds: float = 60.0) -> None:
         self.hang_seconds = hang_seconds
         self.on_ready_calls = 0
 
-    async def on_ready(  # type: ignore[override]
+    @override
+    async def on_ready(
         self,
         source: ScheduleSource,
         task: ScheduledTask,
@@ -50,7 +52,6 @@ def _task() -> ScheduledTask:
     )
 
 
-@pytest.mark.anyio
 async def test_send_with_timeout_returns_on_timeout_without_raising(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
@@ -87,7 +88,6 @@ async def test_send_with_timeout_returns_on_timeout_without_raising(
     assert "dummy" in msg
 
 
-@pytest.mark.anyio
 async def test_send_with_timeout_does_not_log_on_success(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
@@ -99,7 +99,8 @@ def __init__(self) -> None:
             super().__init__(broker=InMemoryBroker(), sources=[_StubSource()])
             self.calls = 0
 
-        async def on_ready(  # type: ignore[override]
+        @override
+        async def on_ready(
             self,
             source: ScheduleSource,
             task: ScheduledTask,
@@ -121,7 +122,6 @@ async def on_ready(  # type: ignore[override]
     assert warnings == []
 
 
-@pytest.mark.anyio
 async def test_send_with_timeout_propagates_non_timeout_exceptions() -> None:
     """Errors inside on_ready that AREN'T a timeout must still propagate.
 
@@ -134,7 +134,8 @@ class _BoomScheduler(TaskiqScheduler):
         def __init__(self) -> None:
             super().__init__(broker=InMemoryBroker(), sources=[_StubSource()])
 
-        async def on_ready(  # type: ignore[override]
+        @override
+        async def on_ready(
             self,
             source: ScheduleSource,
             task: ScheduledTask,
@@ -145,7 +146,6 @@ async def on_ready(  # type: ignore[override]
         await send_with_timeout(_BoomScheduler(), _StubSource(), _task(), timeout=5.0)
 
 
-@pytest.mark.anyio
 async def test_send_with_timeout_cancels_inner_send() -> None:
     """The inner ``send`` coroutine must actually be cancelled on timeout.
 
@@ -158,7 +158,8 @@ class _CancelObservingScheduler(TaskiqScheduler):
         def __init__(self) -> None:
             super().__init__(broker=InMemoryBroker(), sources=[_StubSource()])
 
-        async def on_ready(  # type: ignore[override]
+        @override
+        async def on_ready(
             self,
             source: ScheduleSource,
             task: ScheduledTask,
@@ -179,7 +180,6 @@ async def on_ready(  # type: ignore[override]
     await asyncio.wait_for(cancelled.wait(), timeout=2.0)
 
 
-@pytest.mark.anyio
 async def test_plain_send_still_works_unchanged() -> None:
     """The original ``send`` function must remain unchanged in behavior.
 
@@ -191,7 +191,8 @@ def __init__(self) -> None:
             super().__init__(broker=InMemoryBroker(), sources=[_StubSource()])
             self.calls: list[Any] = []
 
-        async def on_ready(  # type: ignore[override]
+        @override
+        async def on_ready(
             self,
             source: ScheduleSource,
             task: ScheduledTask,