Multi-database client implementation

.github/workflows/install_and_test.sh

@@ -40,7 +40,7 @@ cd ${TESTDIR}
 # install, run tests
 pip install ${PKG}
 # Redis tests
-pytest -m 'not onlycluster' --ignore=tests/test_scenario
+pytest -m 'not onlycluster' --ignore=tests/test_scenario --ignore=tests/test_asyncio/test_scenario
 # RedisCluster tests
 CLUSTER_URL="redis://localhost:16379/0"
 CLUSTER_SSL_URL="rediss://localhost:27379/0"

dev_requirements.txt

@@ -14,3 +14,4 @@ uvloop
 vulture>=2.3.0
 numpy>=1.24.0
 redis-entraid==1.0.0
+pybreaker>=1.4.0

pyproject.toml

@@ -42,6 +42,9 @@ ocsp = [
 jwt = [
     "PyJWT>=2.9.0",
 ]
+circuit_breaker = [
+    "pybreaker>=1.4.0"
+]
 
 [project.urls]
 Changes = "https://github.com/redis/redis-py/releases"

redis/asyncio/client.py

@@ -1239,6 +1239,7 @@ async def run(
         *,
         exception_handler: Optional["PSWorkerThreadExcHandlerT"] = None,
         poll_timeout: float = 1.0,
+        pubsub=None,
     ) -> None:
         """Process pub/sub messages using registered callbacks.
 
@@ -1263,9 +1264,14 @@ async def run(
         await self.connect()
         while True:
             try:
-                await self.get_message(
-                    ignore_subscribe_messages=True, timeout=poll_timeout
-                )
+                if pubsub is None:
+                    await self.get_message(
+                        ignore_subscribe_messages=True, timeout=poll_timeout
+                    )
+                else:
+                    await pubsub.get_message(
+                        ignore_subscribe_messages=True, timeout=poll_timeout
+                    )
             except asyncio.CancelledError:
                 raise
             except BaseException as e:

redis/asyncio/cluster.py

@@ -409,6 +409,7 @@ def __init__(
         else:
             self._event_dispatcher = event_dispatcher
 
+        self.startup_nodes = startup_nodes
         self.nodes_manager = NodesManager(
             startup_nodes,
             require_full_coverage,
@@ -2253,7 +2254,10 @@ async def _reinitialize_on_error(self, error):
                 await self._pipe.cluster_client.nodes_manager.initialize()
                 self.reinitialize_counter = 0
             else:
-                self._pipe.cluster_client.nodes_manager.update_moved_exception(error)
+                if isinstance(error, AskError):
+                    self._pipe.cluster_client.nodes_manager.update_moved_exception(
+                        error
+                    )
 
         self._executing = False
 

redis/asyncio/connection.py

@@ -213,6 +213,7 @@ def __init__(
         self._connect_callbacks: List[weakref.WeakMethod[ConnectCallbackT]] = []
         self._buffer_cutoff = 6000
         self._re_auth_token: Optional[TokenInterface] = None
+        self._should_reconnect = False
 
         try:
             p = int(protocol)
@@ -343,6 +344,12 @@ async def connect_check_health(
             if task and inspect.isawaitable(task):
                 await task
 
+    def mark_for_reconnect(self):
+        self._should_reconnect = True
+
+    def should_reconnect(self):
+        return self._should_reconnect
+
     @abstractmethod
     async def _connect(self):
         pass
@@ -1240,6 +1247,9 @@ async def release(self, connection: AbstractConnection):
         # Connections should always be returned to the correct pool,
         # not doing so is an error that will cause an exception here.
         self._in_use_connections.remove(connection)
+        if connection.should_reconnect():
+            await connection.disconnect()
+
         self._available_connections.append(connection)
         await self._event_dispatcher.dispatch_async(
             AsyncAfterConnectionReleasedEvent(connection)
@@ -1267,6 +1277,14 @@ async def disconnect(self, inuse_connections: bool = True):
         if exc:
             raise exc
 
+    async def update_active_connections_for_reconnect(self):
+        """
+        Mark all active connections for reconnect.
+        """
+        async with self._lock:
+            for conn in self._in_use_connections:
+                conn.mark_for_reconnect()
+
     async def aclose(self) -> None:
         """Close the pool, disconnecting all connections"""
         await self.disconnect()

redis/asyncio/http/http_client.py

@@ -0,0 +1,265 @@
+import asyncio
+from abc import ABC, abstractmethod
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, Mapping, Optional, Union
+
+from redis.http.http_client import HttpClient, HttpResponse
+
+DEFAULT_USER_AGENT = "HttpClient/1.0 (+https://example.invalid)"
+DEFAULT_TIMEOUT = 30.0
+RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
+
+
+class AsyncHTTPClient(ABC):
+    @abstractmethod
+    async def get(
+        self,
+        path: str,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP GET request."""
+        pass
+
+    @abstractmethod
+    async def delete(
+        self,
+        path: str,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP DELETE request."""
+        pass
+
+    @abstractmethod
+    async def post(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP POST request."""
+        pass
+
+    @abstractmethod
+    async def put(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP PUT request."""
+        pass
+
+    @abstractmethod
+    async def patch(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        """
+        Invoke HTTP PATCH request."""
+        pass
+
+    @abstractmethod
+    async def request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        body: Optional[Union[bytes, str]] = None,
+        timeout: Optional[float] = None,
+    ) -> HttpResponse:
+        """
+        Invoke HTTP request with given method."""
+        pass
+
+
+class AsyncHTTPClientWrapper(AsyncHTTPClient):
+    """
+    An async wrapper around sync HTTP client with thread pool execution.
+    """
+
+    def __init__(self, client: HttpClient, max_workers: int = 10) -> None:
+        """
+        Initialize a new HTTP client instance.
+
+        Args:
+            client: Sync HTTP client instance.
+            max_workers: Maximum number of concurrent requests.
+
+        The client supports both regular HTTPS with server verification and mutual TLS
+        authentication. For server verification, provide CA certificate information via
+        ca_file, ca_path or ca_data. For mutual TLS, additionally provide a client
+        certificate and key via client_cert_file and client_key_file.
+        """
+        self.client = client
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+
+    async def get(
+        self,
+        path: str,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor, self.client.get, path, params, headers, timeout, expect_json
+        )
+
+    async def delete(
+        self,
+        path: str,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.delete,
+            path,
+            params,
+            headers,
+            timeout,
+            expect_json,
+        )
+
+    async def post(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.post,
+            path,
+            json_body,
+            data,
+            params,
+            headers,
+            timeout,
+            expect_json,
+        )
+
+    async def put(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.put,
+            path,
+            json_body,
+            data,
+            params,
+            headers,
+            timeout,
+            expect_json,
+        )
+
+    async def patch(
+        self,
+        path: str,
+        json_body: Optional[Any] = None,
+        data: Optional[Union[bytes, str]] = None,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: Optional[float] = None,
+        expect_json: bool = True,
+    ) -> Union[HttpResponse, Any]:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.patch,
+            path,
+            json_body,
+            data,
+            params,
+            headers,
+            timeout,
+            expect_json,
+        )
+
+    async def request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[
+            Mapping[str, Union[None, str, int, float, bool, list, tuple]]
+        ] = None,
+        headers: Optional[Mapping[str, str]] = None,
+        body: Optional[Union[bytes, str]] = None,
+        timeout: Optional[float] = None,
+    ) -> HttpResponse:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(
+            self._executor,
+            self.client.request,
+            method,
+            path,
+            params,
+            headers,
+            body,
+            timeout,
+        )