Merge tag '0.1.2' into develop

0.1.2

Author: s3rius

.flake8

@@ -80,6 +80,14 @@ ignore =
   WPS229,
   ; Found function with too much cognitive complexity
   WPS231,
+  ; Found too deep nesting
+  WPS220,
+  ; Found line with high Jones Complexity
+  WPS221,
+  ; function name should be lowercase
+  N802,
+  ; Do not perform function calls in argument defaults.
+  B008,
 
   ; all init files
   __init__.py:
@@ -99,6 +107,10 @@ per-file-ignores =
   WPS432,
   ; Missing parameter(s) in Docstring
   DAR101,
+  ; Found too short name
+  WPS111,
+  ; Found complex default value
+  WPS404,
 
 exclude =
   ./.git,

docs/examples/introduction/aio_pika_broker.py

@@ -22,6 +22,7 @@ async def main() -> None:
         print(f"Returned value: {result.return_value}")
     else:
         print("Error found while executing task.")
+    await broker.shutdown()
 
 
 if __name__ == "__main__":

docs/examples/introduction/full_example.py

@@ -26,6 +26,7 @@ async def main() -> None:
         print(f"Returned value: {result.return_value}")
     else:
         print("Error found while executing task.")
+    await broker.shutdown()
 
 
 if __name__ == "__main__":

docs/examples/introduction/inmemory_run.py

@@ -12,6 +12,7 @@ async def add_one(value: int) -> int:
 
 
 async def main() -> None:
+    await broker.startup()
     # Send the task to the broker.
     task = await add_one.kiq(1)
     # Wait for the result.
@@ -21,6 +22,7 @@ async def main() -> None:
         print(f"Returned value: {result.return_value}")
     else:
         print("Error found while executing task.")
+    await broker.shutdown()
 
 
 if __name__ == "__main__":

docs/examples/state/async_generator_deps.py

@@ -0,0 +1,18 @@
+import asyncio
+from typing import 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: str = TaskiqDepends(dependency)) -> None:
+    print(dep.upper())

docs/examples/state/class_dependency.py

@@ -0,0 +1,17 @@
+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: str = TaskiqDepends(db_connection)) -> None:
+        self.db_conn = db_conn
+
+    def get_users(self) -> str:
+        return self.db_conn.upper()
+
+
+def my_task(dao: MyDAO = TaskiqDepends()) -> None:
+    print(dao.get_users())

docs/examples/state/dependencies_tree.py

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

docs/examples/state/events_example.py

@@ -0,0 +1,65 @@
+import asyncio
+from typing import 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",
+    result_backend=RedisAsyncResultBackend(
+        "redis://localhost/0",
+    ),
+)
+
+
+@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: 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: 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.py

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

docs/examples/state/no_cache.py

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