Merge branch 'release/0.1.0'

Author: s3rius

README.md

@@ -24,7 +24,7 @@ from taskiq_redis.redis_backend import RedisAsyncResultBackend
 
 
 redis_async_result = RedisAsyncResultBackend(
-    url="redis://localhost:6379",
+    redis_url="redis://localhost:6379",
 )
 
 broker = RedisBroker(
@@ -41,7 +41,7 @@ async def best_task_ever() -> None:
 
 
 async def main():
-    task = await my_async_task.kiq()
+    task = await best_task_ever.kiq()
     print(await task.get_result())
 
 
@@ -60,4 +60,5 @@ RedisBroker parameters:
 ## RedisAsyncResultBackend configuration
 
 RedisAsyncResultBackend parameters:
-* `url` - url to redis.
+* `redis_url` - url to redis.
+* `keep_results` - flag to not remove results from Redis after reading.

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskiq-redis"
-version = "0.0.4"
+version = "0.1.0"
 description = "Redis integration for taskiq"
 authors = ["taskiq-team <taskiq@norely.com>"]
 readme = "README.md"
@@ -19,7 +19,7 @@ keywords = ["taskiq", "tasks", "distributed", "async", "redis", "result_backend"
 
 [tool.poetry.dependencies]
 python = "^3.7"
-taskiq = "^0"
+taskiq = "^0.2.0"
 redis = "^4.2.0"
 
 [tool.poetry.dev-dependencies]

pyproject.toml

@@ -11,8 +11,15 @@
 class RedisAsyncResultBackend(AsyncResultBackend[_ReturnType]):
     """Async result based on redis."""
 
-    def __init__(self, redis_url: str):
+    def __init__(self, redis_url: str, keep_results: bool = True):
+        """
+        Constructs a new result backend.
+
+        :param redis_url: url to redis.
+        :param keep_results: flag to not remove results from Redis after reading.
+        """
         self.redis_pool = ConnectionPool.from_url(redis_url)
+        self.keep_results = keep_results
 
     async def shutdown(self) -> None:
         """Closes redis connection."""
@@ -80,6 +87,9 @@ async def get_result(  # noqa: WPS210
                 keys=fields,
             )
 
+            if not self.keep_results:
+                await redis.delete(task_id)
+
         result = {
             result_key: pickle.loads(result_value)
             for result_value, result_key in zip(result_values, fields)

taskiq_redis/redis_backend.py

@@ -1,7 +1,6 @@
-import asyncio
 import pickle
 from logging import getLogger
-from typing import Any, Callable, Coroutine, Optional, TypeVar
+from typing import Any, AsyncGenerator, Callable, Optional, TypeVar
 
 from redis.asyncio import ConnectionPool, Redis
 from taskiq.abc.broker import AsyncBroker
@@ -70,19 +69,15 @@ async def kick(self, message: BrokerMessage) -> None:
                 pickle.dumps(message),
             )
 
-    async def listen(
-        self,
-        callback: Callable[[BrokerMessage], Coroutine[Any, Any, None]],
-    ) -> None:
+    async def listen(self) -> AsyncGenerator[BrokerMessage, None]:
         """
-        Listen redis list for new messages.
+        Listen redis queue for new messages.
 
-        This function listens to list calls callback on
-        new messages.
+        This function listens to the queue
+        and yields new messages if they have BrokerMessage type.
 
-        :param callback: function to call on new message.
+        :yields: broker messages.
         """
-        loop = asyncio.get_event_loop()
         async with Redis(connection_pool=self.connection_pool) as redis_conn:
             redis_pubsub_channel = redis_conn.pubsub()
             await redis_pubsub_channel.subscribe(self.redis_pubsub_channel)
@@ -93,7 +88,7 @@ async def listen(
                             message["data"],
                         )
                         if isinstance(redis_message, BrokerMessage):
-                            loop.create_task(callback(redis_message))
+                            yield redis_message
                     except (
                         TypeError,
                         AttributeError,

taskiq_redis/redis_broker.py

@@ -68,3 +68,59 @@ async def test_fetch_without_logs(redis_url: str) -> None:
     assert fetched_result.return_value == 11
     assert fetched_result.execution_time == 112.2  # noqa: WPS459
     assert fetched_result.is_err
+
+
+@pytest.mark.anyio
+async def test_remove_results_after_reading(redis_url: str) -> None:
+    """
+    Check if removing results after reading works fine.
+
+    :param redis_url: redis URL.
+    """
+    result_backend = RedisAsyncResultBackend(  # type: ignore
+        redis_url=redis_url,
+        keep_results=False,
+    )
+    task_id = uuid.uuid4().hex
+    result: "TaskiqResult[int]" = TaskiqResult(
+        is_err=True,
+        log="My Log",
+        return_value=11,
+        execution_time=112.2,
+    )
+    await result_backend.set_result(
+        task_id=task_id,
+        result=result,
+    )
+
+    await result_backend.get_result(task_id=task_id)
+    with pytest.raises(Exception):
+        await result_backend.get_result(task_id=task_id)
+
+
+@pytest.mark.anyio
+async def test_keep_results_after_reading(redis_url: str) -> None:
+    """
+    Check if keeping results after reading works fine.
+
+    :param redis_url: redis URL.
+    """
+    result_backend = RedisAsyncResultBackend(  # type: ignore
+        redis_url=redis_url,
+        keep_results=True,
+    )
+    task_id = uuid.uuid4().hex
+    result: "TaskiqResult[int]" = TaskiqResult(
+        is_err=True,
+        log="My Log",
+        return_value=11,
+        execution_time=112.2,
+    )
+    await result_backend.set_result(
+        task_id=task_id,
+        result=result,
+    )
+
+    res1 = await result_backend.get_result(task_id=task_id)
+    res2 = await result_backend.get_result(task_id=task_id)
+    assert res1 == res2