Merge pull request #15 from booqoffsky/issue-14

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.

taskiq_redis/redis_backend.py

@@ -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/tests/test_result_backend.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