Added redis broker.

Author: chandr-andr

.flake8

@@ -78,6 +78,8 @@ ignore =
   ; Consider possible security implications associated with pickle module
   ; Pickle and modules that wrap it can be unsafe when used to deserialize untrusted data, possible security issue
   S403, S301
+  ; Found `%` string formatting
+  WPS323
 
 per-file-ignores =
   ; all tests

poetry.lock

@@ -7,9 +7,9 @@ packages = [{ include = "taskiq", from = "taskiq_redis" }]
 
 [tool.poetry.dependencies]
 python = "^3.7"
-taskiq = { git = "https://github.com/taskiq-python/taskiq", branch = "master" }
 redis = "4.3.4"
 flake8 = "^4.0.1"
+taskiq = {path = "../taskiq"}
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.0"

pyproject.toml

@@ -0,0 +1,106 @@
+import pickle
+from logging import getLogger
+from typing import Any, AsyncGenerator, Callable, Optional, TypeVar
+
+from redis.asyncio import ConnectionPool, Redis
+from taskiq.abc.broker import AsyncBroker
+from taskiq.abc.result_backend import AsyncResultBackend
+from taskiq.message import BrokerMessage
+
+_T = TypeVar("_T")  # noqa: WPS111
+
+logger = getLogger("taskiq.redis_broker")
+
+
+class RedisBroker(AsyncBroker):
+    """Broker that works with Redis."""
+
+    def __init__(
+        self,
+        url: Optional[str] = None,
+        task_id_generator: Optional[Callable[[], str]] = None,
+        result_backend: Optional[AsyncResultBackend[_T]] = None,
+        queue_name: str = "taskiq",
+        max_connection_pool_size: Optional[int] = None,
+        **connection_kwargs: Any,
+    ) -> None:
+        """
+        Constructs a new broker.
+
+        :param url: url to redis.
+        :param task_id_generator: custom task_id generator.
+        :param result_backend: custom result backend.
+        :param queue_name: name for a list in redis.
+        :param max_connection_pool_size: maximum number of connections in pool.
+        :param connection_kwargs: additional arguments for aio-redis ConnectionPool.
+        """
+        super().__init__(
+            result_backend=result_backend,
+            task_id_generator=task_id_generator,
+        )
+
+        self.connection_pool: ConnectionPool = ConnectionPool.from_url(
+            url=url,
+            max_connections=max_connection_pool_size,
+            **connection_kwargs,
+        )
+
+        self.redis_list_name = queue_name
+
+    async def shutdown(self) -> None:
+        """Closes redis connection pool."""
+        self.connection_pool.disconnect()
+
+    async def kick(self, message: BrokerMessage) -> None:
+        """
+        Sends a message to the redis broker list.
+
+        This function constructs message for redis
+        and sends it.
+
+        The message is pickled dict object with message,
+        task_id, task_name and labels.
+
+        :param message: message to send.
+        """
+        async with Redis(connection_pool=self.connection_pool) as redis_conn:
+            await redis_conn.rpush(
+                self.redis_list_name,
+                pickle.dumps(message),
+            )
+
+    async def listen(self) -> AsyncGenerator[BrokerMessage, None]:
+        """
+        Listen redis list for new messages.
+
+        This function listens to list and yields new messages.
+
+        :yields: parsed broker messages.
+        """
+        async with Redis(connection_pool=self.connection_pool) as redis_conn:
+            while True:
+                redis_pickled_message = await redis_conn.blpop(
+                    self.redis_list_name,
+                )
+                if redis_pickled_message:
+                    try:  # noqa: WPS448
+                        redis_message = pickle.loads(  # noqa: WPS220
+                            redis_pickled_message[1],
+                        )
+                        yield BrokerMessage(  # noqa: WPS220
+                            task_id=redis_message.task_id,
+                            task_name=redis_message.task_name,
+                            message=redis_message.message,
+                            labels=redis_message.labels,
+                        )
+                    except (
+                        ValueError,
+                        LookupError,
+                        AttributeError,
+                        pickle.UnpicklingError,
+                    ) as exc:
+                        logger.debug(  # noqa: WPS220
+                            "Cannot read broker message %s",
+                            exc,
+                            exc_info=True,
+                        )