Merge branch 'release/0.9.0'

Author: s3rius

docs/available-components/schedule-sources.md

@@ -17,7 +17,9 @@ The format of the schedule label is the following:
 @broker.task(
     schedule=[
         {
-            "cron": "* * * * *", # type: str, required argument.
+            "cron": "* * * * *", # type: str, either cron or time shoule be specified.
+            "cron_offset": None # type: str | timedelta | None, can be ommited.
+            "time": None  # type: datetime | None, either cron or time shoule be specified.
             "args": [], # type List[Any] | None, can be omitted.
             "kwargs": {}, # type: Dict[str, Any] | None, can be omitted.
             "labels": {}, # type: Dict[str, Any] | None, can be omitted.
@@ -31,11 +33,13 @@ async def my_task():
 Parameters:
 
 - `cron` - crontab string when to run the task.
+- `cron_offset` - timezone offset for cron values. Explained [here](../guide/scheduling-tasks.md#working-with-timezones)
+- `time` - specific time when send the task.
 - `args` - args to use, when invoking the task.
 - `kwargs` - key-word arguments to use when invoking the task.
 - `labels` - additional labels to use when invoking the task.
 
-Usage:
+To enable this source, just add it to the list of sources:
 
 ```python
 from taskiq.scheduler import TaskiqScheduler

docs/examples/state/async_generator_deps_annot.py

@@ -0,0 +1,18 @@
+import asyncio
+from typing import Annotated, AsyncGenerator
+
+from taskiq import TaskiqDepends
+
+
+async def dependency() -> AsyncGenerator[str, None]:
+    print("Startup")
+    await asyncio.sleep(0.1)
+
+    yield "value"
+
+    await asyncio.sleep(0.1)
+    print("Shutdown")
+
+
+async def my_task(dep: Annotated[str, TaskiqDepends(dependency)]) -> None:
+    print(dep.upper())

docs/examples/state/class_dependency_annot.py

@@ -0,0 +1,19 @@
+from typing import Annotated
+
+from taskiq import TaskiqDepends
+
+
+async def db_connection() -> str:
+    return "let's pretend as this is a connection"
+
+
+class MyDAO:
+    def __init__(self, db_conn: Annotated[str, TaskiqDepends(db_connection)]) -> None:
+        self.db_conn = db_conn
+
+    def get_users(self) -> str:
+        return self.db_conn.upper()
+
+
+def my_task(dao: Annotated[MyDAO, TaskiqDepends()]) -> None:
+    print(dao.get_users())

docs/examples/state/dependencies_tree_annot.py

@@ -0,0 +1,27 @@
+import random
+from typing import Annotated
+
+from taskiq import TaskiqDepends
+
+
+def common_dep() -> int:
+    # For example it returns 8
+    return random.randint(1, 10)
+
+
+def dep1(cd: Annotated[int, TaskiqDepends(common_dep)]) -> int:
+    # This function will return 9
+    return cd + 1
+
+
+def dep2(cd: Annotated[int, TaskiqDepends(common_dep)]) -> int:
+    # This function will return 10
+    return cd + 2
+
+
+def my_task(
+    d1: Annotated[int, TaskiqDepends(dep1)],
+    d2: Annotated[int, TaskiqDepends(dep2)],
+) -> int:
+    # This function will return 19
+    return d1 + d2

docs/examples/state/events_example_annot.py

@@ -0,0 +1,69 @@
+import asyncio
+from typing import Annotated, Optional
+
+from redis.asyncio import ConnectionPool, Redis  # type: ignore
+from taskiq_aio_pika import AioPikaBroker
+from taskiq_redis import RedisAsyncResultBackend
+
+from taskiq import Context, TaskiqDepends, TaskiqEvents, TaskiqState
+
+# To run this example, please install:
+# * taskiq
+# * taskiq-redis
+# * taskiq-aio-pika
+
+broker = AioPikaBroker(
+    "amqp://localhost",
+).with_result_backend(RedisAsyncResultBackend("redis://localhost"))
+
+
+@broker.on_event(TaskiqEvents.WORKER_STARTUP)
+async def startup(state: TaskiqState) -> None:
+    # Here we store connection pool on startup for later use.
+    state.redis = ConnectionPool.from_url("redis://localhost/1")
+
+
+@broker.on_event(TaskiqEvents.WORKER_SHUTDOWN)
+async def shutdown(state: TaskiqState) -> None:
+    # Here we close our pool on shutdown event.
+    await state.redis.disconnect()
+
+
+@broker.task
+async def get_val(
+    key: str,
+    context: Annotated[Context, TaskiqDepends()],
+) -> Optional[str]:
+    # Now we can use our pool.
+    redis = Redis(connection_pool=context.state.redis, decode_responses=True)
+    return await redis.get(key)
+
+
+@broker.task
+async def set_val(
+    key: str,
+    value: str,
+    context: Annotated[Context, TaskiqDepends()],
+) -> None:
+    # Now we can use our pool to set value.
+    await Redis(connection_pool=context.state.redis).set(key, value)
+
+
+async def main() -> None:
+    await broker.startup()
+
+    set_task = await set_val.kiq("key", "value")
+    set_result = await set_task.wait_result(with_logs=True)
+    if set_result.is_err:
+        print(set_result.log)
+        raise ValueError("Cannot set value in redis. See logs.")
+
+    get_task = await get_val.kiq("key")
+    get_res = await get_task.wait_result()
+    print(f"Got redis value: {get_res.return_value}")
+
+    await broker.shutdown()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

docs/examples/state/generator_deps_annot.py

@@ -0,0 +1,15 @@
+from typing import Annotated, Generator
+
+from taskiq import TaskiqDepends
+
+
+def dependency() -> Generator[str, None, None]:
+    print("Startup")
+
+    yield "value"
+
+    print("Shutdown")
+
+
+async def my_task(dep: Annotated[str, TaskiqDepends(dependency)]) -> None:
+    print(dep.upper())

docs/examples/state/generics.py

@@ -0,0 +1,27 @@
+from typing import Generic, TypeVar
+
+from taskiq import TaskiqDepends
+
+T = TypeVar("T")
+
+
+def default_dependency() -> int:
+    return 1
+
+
+class GenericDep(Generic[T]):
+    def __init__(self, dep: T = TaskiqDepends()) -> None:
+        self.dep = dep
+
+    def print_dep(self) -> None:
+        print(self.dep)
+
+
+class MyClassDep:
+    def __init__(self, i: int = TaskiqDepends(default_dependency)) -> None:
+        self.i = i
+
+
+async def my_task(dep: GenericDep[MyClassDep] = TaskiqDepends()) -> None:
+    assert isinstance(dep.dep, MyClassDep)
+    dep.print_dep()

docs/examples/state/generics_annot.py

@@ -0,0 +1,27 @@
+from typing import Annotated, Generic, TypeVar
+
+from taskiq import TaskiqDepends
+
+T = TypeVar("T")
+
+
+def default_dependency() -> int:
+    return 1
+
+
+class GenericDep(Generic[T]):
+    def __init__(self, dep: Annotated[T, TaskiqDepends()]) -> None:
+        self.dep = dep
+
+    def print_dep(self) -> None:
+        print(self.dep)
+
+
+class MyClassDep:
+    def __init__(self, i: Annotated[int, TaskiqDepends(default_dependency)]) -> None:
+        self.i = i
+
+
+async def my_task(dep: Annotated[GenericDep[MyClassDep], TaskiqDepends()]) -> None:
+    assert isinstance(dep.dep, MyClassDep)
+    dep.print_dep()

docs/examples/state/no_cache_annot.py

@@ -0,0 +1,23 @@
+import random
+from typing import Annotated
+
+from taskiq import TaskiqDepends
+
+
+def common_dep() -> int:
+    return random.randint(1, 10)
+
+
+def dep1(cd: Annotated[int, TaskiqDepends(common_dep)]) -> int:
+    return cd + 1
+
+
+def dep2(cd: Annotated[int, TaskiqDepends(common_dep, use_cache=False)]) -> int:
+    return cd + 2
+
+
+def my_task(
+    d1: Annotated[int, TaskiqDepends(dep1)],
+    d2: Annotated[int, TaskiqDepends(dep2)],
+) -> int:
+    return d1 + d2

docs/framework_integrations/taskiq-with-aiohttp.md

@@ -60,6 +60,24 @@ taskiq_aiohttp.init(broker, "my_project.main:app")
 From this point, you'll be able to reuse the same dependencies as with `aiohttp-deps`.
 Let's take a look at this function:
 
+::: tabs
+
+@tab Annotated 3.10+
+
+```python
+from aiohttp import web
+from typing import Annotated
+from taskiq import TaskiqDepends
+from my_project.tkq import broker
+
+@broker.task
+async def my_task(app: Annotated[web.Application, TaskiqDepends()]):
+    ...
+
+```
+
+@tab default values
+
 ```python
 from aiohttp import web
 from taskiq import TaskiqDepends
@@ -71,6 +89,8 @@ async def my_task(app: web.Application = TaskiqDepends()):
 
 ```
 
+:::
+
 In this example, we depend on the current application. We can use its state in a current task or any other dependency. We can take db_pool from your application's state, which is the same pool, as the one you've created on AiohTTP's startup.
 But this application is only a mock of your application. It has correct types and all your variables that you filled on startup, but it doesn't handle any request.
 This integration adds two main dependencies: