Merge branch 'release/0.3.0'

Author: s3rius

README.md

@@ -20,8 +20,7 @@ Let's see the example with the redis broker and redis async result:
 ```python
 import asyncio
 
-from taskiq_redis.redis_broker import ListQueueBroker
-from taskiq_redis.redis_backend import RedisAsyncResultBackend
+from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend
 
 redis_async_result = RedisAsyncResultBackend(
     redis_url="redis://localhost:6379",
@@ -71,9 +70,10 @@ Brokers parameters:
 RedisAsyncResultBackend parameters:
 * `redis_url` - url to redis.
 * `keep_results` - flag to not remove results from Redis after reading.
-* `result_ex_time` - expire time in seconds (by default - 1 minute)
+* `result_ex_time` - expire time in seconds (by default - not specified)
 * `result_px_time` - expire time in milliseconds (by default - not specified)
-> IMPORTANT: You must specify either `result_ex_time` or `result_px_time`.  
+> IMPORTANT: **It is highly recommended to use expire time ​​in RedisAsyncResultBackend**  
+> If you want to add expiration, either `result_ex_time` or `result_px_time` must be set.  
 >```python
 ># First variant
 >redis_async_result = RedisAsyncResultBackend(

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskiq-redis"
-version = "0.2.1"
+version = "0.3.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.2.0"
+taskiq = "^0"
 redis = "^4.2.0"
 
 [tool.poetry.dev-dependencies]

pyproject.toml

@@ -1,5 +1,9 @@
 """Package for redis integration."""
 from taskiq_redis.redis_backend import RedisAsyncResultBackend
-from taskiq_redis.redis_broker import ListQueueBroker
+from taskiq_redis.redis_broker import ListQueueBroker, PubSubBroker
 
-__all__ = ["RedisAsyncResultBackend", "ListQueueBroker"]
+__all__ = [
+    "RedisAsyncResultBackend",
+    "ListQueueBroker",
+    "PubSubBroker",
+]

taskiq_redis/__init__.py

@@ -41,7 +41,13 @@ def __init__(
         self.result_ex_time = result_ex_time
         self.result_px_time = result_px_time
 
-        if self.result_ex_time == 0 or self.result_px_time == 0:
+        unavailable_conditions = any(
+            (
+                self.result_ex_time is not None and self.result_ex_time <= 0,
+                self.result_px_time is not None and self.result_px_time <= 0,
+            ),
+        )
+        if unavailable_conditions:
             raise ExpireTimeMustBeMoreThanZeroError(
                 "You must select one expire time param and it must be more than zero.",
             )
@@ -51,9 +57,6 @@ def __init__(
                 "Choose either result_ex_time or result_px_time.",
             )
 
-        if not self.result_ex_time and not self.result_px_time:
-            self.result_ex_time = 60
-
     async def shutdown(self) -> None:
         """Closes redis connection."""
         await self.redis_pool.disconnect()

taskiq_redis/redis_backend.py

@@ -1,5 +1,3 @@
-import pickle
-from abc import abstractmethod
 from logging import getLogger
 from typing import Any, AsyncGenerator, Callable, Optional, TypeVar
 
@@ -49,68 +47,66 @@ def __init__(
 
     async def shutdown(self) -> None:
         """Closes redis connection pool."""
+        await super().shutdown()
         await self.connection_pool.disconnect()
 
-    async def listen(self) -> AsyncGenerator[BrokerMessage, None]:
-        """
-        Listen redis queue for new messages.
 
-        This function listens to the queue
-        and yields new messages if they have BrokerMessage type.
+class PubSubBroker(BaseRedisBroker):
+    """Broker that works with Redis and broadcasts tasks to all workers."""
 
-        :yields: broker messages.
-        """
-        async for message in self._listen_to_raw_messages():
-            try:
-                redis_message = pickle.loads(message)
-                if isinstance(redis_message, BrokerMessage):
-                    yield redis_message
-            except (
-                TypeError,
-                AttributeError,
-                pickle.UnpicklingError,
-            ) as exc:
-                logger.debug(
-                    "Cannot read broker message %s",
-                    exc,
-                    exc_info=True,
-                )
-
-    @abstractmethod
-    async def _listen_to_raw_messages(self) -> AsyncGenerator[bytes, None]:
+    async def kick(self, message: BrokerMessage) -> None:
         """
-        Generator for reading raw data from Redis.
+        Publish message over PUBSUB channel.
 
-        :yields: raw data.
+        :param message: message to send.
         """
-        yield  # type: ignore
-
+        async with Redis(connection_pool=self.connection_pool) as redis_conn:
+            await redis_conn.publish(self.queue_name, message.message)
 
-class PubSubBroker(BaseRedisBroker):
-    """Broker that works with Redis and broadcasts tasks to all workers."""
+    async def listen(self) -> AsyncGenerator[bytes, None]:
+        """
+        Listen redis queue for new messages.
 
-    async def kick(self, message: BrokerMessage) -> None:  # noqa: D102
-        async with Redis(connection_pool=self.connection_pool) as redis_conn:
-            await redis_conn.publish(self.queue_name, pickle.dumps(message))
+        This function listens to the pubsub channel
+        and yields all messages with proper types.
 
-    async def _listen_to_raw_messages(self) -> AsyncGenerator[bytes, None]:
+        :yields: broker messages.
+        """
         async with Redis(connection_pool=self.connection_pool) as redis_conn:
             redis_pubsub_channel = redis_conn.pubsub()
             await redis_pubsub_channel.subscribe(self.queue_name)
             async for message in redis_pubsub_channel.listen():
                 if not message:
                     continue
+                if message["type"] != "message":
+                    logger.debug("Received non-message from redis: %s", message)
+                    continue
                 yield message["data"]
 
 
 class ListQueueBroker(BaseRedisBroker):
     """Broker that works with Redis and distributes tasks between workers."""
 
-    async def kick(self, message: BrokerMessage) -> None:  # noqa: D102
+    async def kick(self, message: BrokerMessage) -> None:
+        """
+        Put a message in a list.
+
+        This method appends a message to the list of all messages.
+
+        :param message: message to append.
+        """
         async with Redis(connection_pool=self.connection_pool) as redis_conn:
-            await redis_conn.lpush(self.queue_name, pickle.dumps(message))
+            await redis_conn.lpush(self.queue_name, message.message)
 
-    async def _listen_to_raw_messages(self) -> AsyncGenerator[bytes, None]:
+    async def listen(self) -> AsyncGenerator[bytes, None]:
+        """
+        Listen redis queue for new messages.
+
+        This function listens to the queue
+        and yields new messages if they have BrokerMessage type.
+
+        :yields: broker messages.
+        """
         redis_brpop_data_position = 1
         async with Redis(connection_pool=self.connection_pool) as redis_conn:
             while True:  # noqa: WPS457

taskiq_redis/redis_broker.py

@@ -5,11 +5,11 @@
 import pytest
 from taskiq import TaskiqResult
 
+from taskiq_redis import RedisAsyncResultBackend
 from taskiq_redis.exceptions import (
     DuplicateExpireTimeSelectedError,
     ExpireTimeMustBeMoreThanZeroError,
 )
-from taskiq_redis.redis_backend import RedisAsyncResultBackend
 
 _ReturnType = TypeVar("_ReturnType")
 
@@ -132,7 +132,13 @@ async def test_cant_specify_ex_and_px_params(
 
 
 @pytest.mark.anyio
+@pytest.mark.parametrize(
+    "ex_time, px_time",
+    [(0, 0), (-500, 0), (0, -500), (-500, -500)],
+)
 async def test_ex_or_px_must_be_more_than_zero(
+    ex_time: int,
+    px_time: int,
     redis_url: str,
 ) -> None:
     """
@@ -143,8 +149,8 @@ async def test_ex_or_px_must_be_more_than_zero(
     with pytest.raises(ExpireTimeMustBeMoreThanZeroError):
         RedisAsyncResultBackend(
             redis_url,
-            result_ex_time=0,
-            result_px_time=0,
+            result_ex_time=ex_time,
+            result_px_time=px_time,
         )
 
 

tests/test_backend.py

@@ -4,10 +4,10 @@
 import pytest
 from taskiq import AsyncBroker, BrokerMessage
 
-from taskiq_redis.redis_broker import ListQueueBroker, PubSubBroker
+from taskiq_redis import ListQueueBroker, PubSubBroker
 
 
-async def get_message(broker: AsyncBroker) -> BrokerMessage:  # type: ignore
+async def get_message(broker: AsyncBroker) -> bytes:  # type: ignore
     """
     Get a message from the broker.
 
@@ -56,7 +56,7 @@ async def test_pub_sub_broker(
 
     message1 = worker1_task.result()
     message2 = worker2_task.result()
-    assert message1 == valid_broker_message
+    assert message1 == valid_broker_message.message
     assert message1 == message2
 
 
@@ -81,4 +81,4 @@ async def test_list_queue_broker(
 
     assert worker1_task.done() != worker2_task.done()
     message = worker1_task.result() if worker1_task.done() else worker2_task.result()
-    assert message == valid_broker_message
+    assert message == valid_broker_message.message

tests/test_broker.py

@@ -3,7 +3,7 @@
 import pytest
 from taskiq import TaskiqResult
 
-from taskiq_redis.redis_backend import RedisAsyncResultBackend
+from taskiq_redis import RedisAsyncResultBackend
 
 
 @pytest.mark.anyio