Fixed linter errors and asyncio use in test_cluster()

Author: agronholm

redis/anyio/_commands/cluster.py

@@ -10,10 +10,11 @@
     Optional,
 )
 
-from ..utils import gather
 from ...commands.cluster import (
-    ClusterDataAccessCommands, ClusterManagementCommands,
-    ClusterMultiKeyCommands, READ_COMMANDS,
+    READ_COMMANDS,
+    ClusterDataAccessCommands,
+    ClusterManagementCommands,
+    ClusterMultiKeyCommands,
 )
 from ...commands.core import (
     AsyncACLCommands,
@@ -25,7 +26,8 @@
 )
 from ...commands.helpers import list_or_args
 from ...commands.redismodules import AsyncRedisModuleCommands
-from ...typing import AnyKeyT, EncodableT, KeyT, KeysT, PatternT
+from ...typing import AnyKeyT, EncodableT, KeysT, KeyT, PatternT
+from ..utils import gather
 
 
 class AsyncClusterMultiKeyCommands(ClusterMultiKeyCommands):
@@ -125,10 +127,7 @@ async def cluster_delslots(self, *slots: EncodableT) -> Sequence[bool]:
         For more information see https://redis.io/commands/cluster-delslots
         """
         return await gather(
-            *(
-                self.execute_command("CLUSTER DELSLOTS", slot)
-                for slot in slots
-            )
+            *(self.execute_command("CLUSTER DELSLOTS", slot) for slot in slots)
         )
 
 

redis/anyio/_parsers/base.py

@@ -7,9 +7,10 @@
 from anyio.abc import ByteStream
 from anyio.streams.buffered import BufferedByteReceiveStream
 
-from redis.exceptions import RedisError, ResponseError
 from redis._parsers import BaseParser, Encoder
+from redis.exceptions import RedisError, ResponseError
 from redis.typing import EncodableT
+
 from ..._parsers.socket import SERVER_CLOSED_CONNECTION_ERROR
 
 

redis/anyio/_parsers/hiredis.py

@@ -5,14 +5,13 @@
 
 from anyio import BrokenResourceError, EndOfStream, move_on_after
 
+from ..._parsers.socket import (
+    SERVER_CLOSED_CONNECTION_ERROR,
+)
 from ...exceptions import ConnectionError, InvalidResponse, RedisError
 from ...typing import EncodableT
 from ...utils import HIREDIS_AVAILABLE
 from .base import AnyIOBaseParser
-from ..._parsers.socket import (
-    SENTINEL,
-    SERVER_CLOSED_CONNECTION_ERROR,
-)
 
 # Used to signal that hiredis-py does not have enough data to parse.
 # Using `False` or `None` is not reliable, given that the parser can

redis/anyio/_parsers/resp2.py

@@ -2,10 +2,10 @@
 
 from typing import Any
 
+from ..._parsers.socket import SERVER_CLOSED_CONNECTION_ERROR
 from ...exceptions import ConnectionError, InvalidResponse, ResponseError
 from ...typing import EncodableT
 from .base import _AnyIORESPBase
-from ..._parsers.socket import SERVER_CLOSED_CONNECTION_ERROR
 
 
 class _AnyIORESP2Parser(_AnyIORESPBase):

redis/anyio/_parsers/resp3.py

@@ -3,10 +3,10 @@
 from logging import getLogger
 from typing import Any
 
+from ..._parsers.socket import SERVER_CLOSED_CONNECTION_ERROR
 from ...exceptions import ConnectionError, InvalidResponse, ResponseError
 from ...typing import EncodableT
 from .base import _AnyIORESPBase
-from ..._parsers.socket import SERVER_CLOSED_CONNECTION_ERROR
 
 _INVALIDATION_MESSAGE = [b"invalidate", "invalidate"]
 

redis/anyio/client.py

@@ -987,10 +987,7 @@ async def check_health(self):
                 "did you forget to call subscribe() or psubscribe()?"
             )
 
-        if (
-            conn.health_check_interval
-            and anyio.current_time() > conn.next_health_check
-        ):
+        if conn.health_check_interval and anyio.current_time() > conn.next_health_check:
             await conn.send_command(
                 "PING", self.HEALTH_CHECK_MESSAGE, check_health=False
             )

redis/anyio/cluster.py

@@ -12,7 +12,6 @@
     Callable,
     Deque,
     Dict,
-    Generator,
     List,
     Mapping,
     Optional,
@@ -461,7 +460,7 @@ async def __aexit__(
         self,
         exc_type: type[BaseException] | None,
         exc_value: BaseException | None,
-        exc_tb: TracebackType | None
+        exc_tb: TracebackType | None,
     ) -> None:
         exit_stack, self._exit_stack = self._exit_stack, None
         await exit_stack.__aexit__(exc_type, exc_value, exc_tb)
@@ -471,7 +470,9 @@ def __del__(
         _warn: Any = warnings.warn,
     ) -> None:
         if getattr(self, "_exit_stack", None) is not None:
-            _warn(f"Unclosed RedisCluster client {self!r}", ResourceWarning, source=self)
+            _warn(
+                f"Unclosed RedisCluster client {self!r}", ResourceWarning, source=self
+            )
 
     async def on_connect(self, connection: Connection) -> None:
         await connection.on_connect()
@@ -1428,7 +1429,7 @@ async def __aexit__(
         self,
         exc_type: type[BaseException] | None,
         exc_value: BaseException | None,
-        exc_tb: TracebackType | None
+        exc_tb: TracebackType | None,
     ) -> bool | None:
         try:
             return await self._exit_stack.__aexit__(exc_type, exc_value, exc_tb)
@@ -1446,12 +1447,7 @@ async def __aexit__(
 
     async def aclose(self, attr: str = "nodes_cache") -> None:
         self.default_node = None
-        await gather(
-            *(
-                node.disconnect()
-                for node in getattr(self, attr).values()
-            )
-        )
+        await gather(*(node.disconnect() for node in getattr(self, attr).values()))
 
     def remap_host_port(self, host: str, port: int) -> Tuple[str, int]:
         """
@@ -1632,10 +1628,7 @@ async def _execute(
             nodes[node.name][1].append(cmd)
 
         errors = await gather(
-            *(
-                node[0].execute_pipeline(node[1])
-                for node in nodes.values()
-            )
+            *(node[0].execute_pipeline(node[1]) for node in nodes.values())
         )
 
         if any(errors):

redis/anyio/connection.py

@@ -31,9 +31,9 @@
 from anyio import BrokenResourceError, aclose_forcefully
 from anyio.abc import SocketAttribute, SocketStream
 
+from ..utils import SSL_AVAILABLE
 from ._parsers.base import AnyIOBaseParser
 from .utils import wait_for_condition
-from ..utils import SSL_AVAILABLE
 
 if SSL_AVAILABLE:
     import ssl
@@ -43,10 +43,6 @@
     TLSVersion = None
     SSLContext = None
 
-from ..auth.token import TokenInterface
-from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
-from ..utils import deprecated_args, format_error_message
-
 from redis.anyio.retry import Retry
 from redis.backoff import NoBackoff
 from redis.connection import DEFAULT_RESP_VERSION
@@ -64,9 +60,11 @@
 from redis.utils import HIREDIS_AVAILABLE, get_lib_version, str_if_bytes
 
 from .._parsers import (
-    BaseParser,
     Encoder,
 )
+from ..auth.token import TokenInterface
+from ..event import AsyncAfterConnectionReleasedEvent, EventDispatcher
+from ..utils import deprecated_args, format_error_message
 from ._parsers import (
     _AnyIOHiredisParser,
     _AnyIORESP2Parser,
@@ -234,9 +232,7 @@ def __del__(
         # collected and therefore produce no resource warnings.  We add one
         # here, in the same style as those from the stdlib.
         if getattr(self, "_stream", None):
-            _warn(
-                f"unclosed Connection {self!r}", ResourceWarning, source=self
-            )
+            _warn(f"unclosed Connection {self!r}", ResourceWarning, source=self)
 
     def __repr__(self):
         repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
@@ -488,10 +484,7 @@ async def _ping_failed(self, error):
 
     async def check_health(self):
         """Check the health of the connection with a PING/PONG"""
-        if (
-            self.health_check_interval
-            and anyio.current_time() > self.next_health_check
-        ):
+        if self.health_check_interval and anyio.current_time() > self.next_health_check:
             await self.retry.call_with_retry(self._send_ping, self._ping_failed)
 
     async def _send_packed_command(self, command: Iterable[bytes]) -> None:

redis/anyio/utils.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Coroutine, Sequence
-from typing import Any, TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 import anyio
 

redis/commands/core.py

@@ -53,8 +53,8 @@
 from .helpers import list_or_args
 
 if TYPE_CHECKING:
-    import redis.asyncio.client
     import redis.anyio.client
+    import redis.asyncio.client
     import redis.client
 
 
@@ -1453,7 +1453,7 @@ def __init__(
         client: Union[
             "redis.client.Redis",
             "redis.asyncio.client.Redis",
-            "redis.anyio.client.Redis"
+            "redis.anyio.client.Redis",
         ],
         key: str,
         default_overflow: Union[str, None] = None,
@@ -1601,7 +1601,7 @@ def bitfield(
         self: Union[
             "redis.client.Redis",
             "redis.asyncio.client.Redis",
-            "redis.anyio.client.Redis"
+            "redis.anyio.client.Redis",
         ],
         key: KeyT,
         default_overflow: Union[str, None] = None,
@@ -1618,7 +1618,7 @@ def bitfield_ro(
         self: Union[
             "redis.client.Redis",
             "redis.asyncio.client.Redis",
-            "redis.anyio.client.Redis"
+            "redis.anyio.client.Redis",
         ],
         key: KeyT,
         encoding: str,
@@ -5706,7 +5706,9 @@ async def __call__(
         self,
         keys: Union[Sequence[KeyT], None] = None,
         args: Union[Iterable[EncodableT], None] = None,
-        client: Union["redis.asyncio.client.Redis", "redis.anyio.client.Redis", None] = None,
+        client: Union[
+            "redis.asyncio.client.Redis", "redis.anyio.client.Redis", None
+        ] = None,
     ):
         """Execute the script, passing any required ``args``"""
         keys = keys or []
@@ -5715,8 +5717,8 @@ async def __call__(
             client = self.registered_client
         args = tuple(keys) + tuple(args)
         # make sure the Redis server knows about the script
-        from redis.asyncio.client import Pipeline as AsyncioPipeline
         from redis.anyio.client import Pipeline as AnyioPipeline
+        from redis.asyncio.client import Pipeline as AsyncioPipeline
 
         if isinstance(client, (AsyncioPipeline, AnyioPipeline)):
             # Make sure the pipeline can register the script before executing.