FIX REVIEW

Author: Takuma Iwaki

channels_redis/core.py

@@ -19,12 +19,12 @@
     create_pool,
     decode_hosts,
 )
-from typing import TYPE_CHECKING, Dict, List, Tuple, Union, Optional
 
-if TYPE_CHECKING:
+import typing
+
+if typing.TYPE_CHECKING:
     from redis.asyncio.connection import ConnectionPool
     from redis.asyncio.client import Redis
-    from .core import RedisChannelLayer
     from typing_extensions import Buffer
 
 logger = logging.getLogger(__name__)
@@ -39,10 +39,10 @@ class ChannelLock:
     """
 
     def __init__(self):
-        self.locks: "collections.defaultdict[str, asyncio.Lock]" = (
+        self.locks: collections.defaultdict[str, asyncio.Lock] = (
             collections.defaultdict(asyncio.Lock)
         )
-        self.wait_counts: "collections.defaultdict[str, int]" = collections.defaultdict(
+        self.wait_counts: collections.defaultdict[str, int] = collections.defaultdict(
             int
         )
 
@@ -87,7 +87,7 @@ class RedisLoopLayer:
     def __init__(self, channel_layer: "RedisChannelLayer"):
         self._lock = asyncio.Lock()
         self.channel_layer = channel_layer
-        self._connections: "Dict[int, Redis]" = {}
+        self._connections: typing.Dict[int, "Redis"] = {}
 
     def get_connection(self, index: int) -> "Redis":
         if index not in self._connections:
@@ -145,7 +145,7 @@ def __init__(
             symmetric_encryption_keys=symmetric_encryption_keys,
         )
         # Cached redis connection pools and the event loop they are from
-        self._layers: "Dict[asyncio.AbstractEventLoop, RedisLoopLayer]" = {}
+        self._layers: typing.Dict[asyncio.AbstractEventLoop, "RedisLoopLayer"] = {}
         # Normal channels choose a host index by cycling through the available hosts
         self._receive_index_generator = itertools.cycle(range(len(self.hosts)))
         self._send_index_generator = itertools.cycle(range(len(self.hosts)))
@@ -154,15 +154,15 @@ def __init__(
         # Number of coroutines trying to receive right now
         self.receive_count = 0
         # The receive lock
-        self.receive_lock: "Optional[asyncio.Lock]" = None
+        self.receive_lock: typing.Optional[asyncio.Lock] = None
         # Event loop they are trying to receive on
-        self.receive_event_loop: "Optional[asyncio.AbstractEventLoop]" = None
+        self.receive_event_loop: typing.Optional[asyncio.AbstractEventLoop] = None
         # Buffered messages by process-local channel name
-        self.receive_buffer: "collections.defaultdict[str, BoundedQueue]" = (
+        self.receive_buffer: collections.defaultdict[str, BoundedQueue] = (
             collections.defaultdict(functools.partial(BoundedQueue, self.capacity))
         )
         # Detached channel cleanup tasks
-        self.receive_cleaners: "List[asyncio.Task]" = []
+        self.receive_cleaners: typing.List[asyncio.Task] = []
         # Per-channel cleanup locks to prevent a receive starting and moving
         # a message back into the main queue before its cleanup has completed
         self.receive_clean_locks = ChannelLock()
@@ -180,7 +180,7 @@ async def send(self, channel: str, message):
         """
         # Typecheck
         assert isinstance(message, dict), "message is not a dict"
-        assert self.require_valid_channel_name(channel), "Channel name not valid"
+        assert self.valid_channel_name(channel), "Channel name not valid"
         # Make sure the message does not contain reserved keys
         assert "__asgi_channel__" not in message
         # If it's a process-local channel, strip off local part and stick full name in message
@@ -221,7 +221,7 @@ def _backup_channel_name(self, channel: str) -> str:
         return channel + "$inflight"
 
     async def _brpop_with_clean(
-        self, index: int, channel: str, timeout: "Union[int, float, bytes, str]"
+        self, index: int, channel: str, timeout: typing.Union[int, float, bytes, str]
     ):
         """
         Perform a Redis BRPOP and manage the backup processing queue.
@@ -269,7 +269,7 @@ async def receive(self, channel: str):
         """
         # Make sure the channel name is valid then get the non-local part
         # and thus its index
-        assert self.require_valid_channel_name(channel)
+        assert self.valid_channel_name(channel)
         if "!" in channel:
             real_channel = self.non_local_name(channel)
             assert real_channel.endswith(
@@ -385,14 +385,12 @@ async def receive(self, channel: str):
             # Do a plain direct receive
             return (await self.receive_single(channel))[1]
 
-    async def receive_single(self, channel: str) -> "Tuple":
+    async def receive_single(self, channel: str) -> typing.Tuple:
         """
         Receives a single message off of the channel and returns it.
         """
         # Check channel name
-        assert self.require_valid_channel_name(
-            channel, receive=True
-        ), "Channel name invalid"
+        assert self.valid_channel_name(channel, receive=True), "Channel name invalid"
         # Work out the connection to use
         if "!" in channel:
             assert channel.endswith("!")
@@ -423,7 +421,7 @@ async def receive_single(self, channel: str) -> "Tuple":
             )
             self.receive_cleaners.append(cleaner)
 
-            def _cleanup_done(cleaner: "asyncio.Task"):
+            def _cleanup_done(cleaner: asyncio.Task):
                 self.receive_cleaners.remove(cleaner)
                 self.receive_clean_locks.release(channel_key)
 
@@ -497,8 +495,8 @@ async def group_add(self, group: str, channel: str):
         Adds the channel name to a group.
         """
         # Check the inputs
-        assert self.require_valid_group_name(group), True
-        assert self.require_valid_channel_name(channel), True
+        assert self.valid_group_name(group), True
+        assert self.valid_channel_name(channel), True
         # Get a connection to the right shard
         group_key = self._group_key(group)
         connection = self.connection(self.consistent_hash(group))
@@ -513,8 +511,8 @@ async def group_discard(self, group: str, channel: str):
         Removes the channel from the named group if it is in the group;
         does nothing otherwise (does not error)
         """
-        assert self.require_valid_group_name(group), "Group name not valid"
-        assert self.require_valid_channel_name(channel), "Channel name not valid"
+        assert self.valid_group_name(group), "Group name not valid"
+        assert self.valid_channel_name(channel), "Channel name not valid"
         key = self._group_key(group)
         connection = self.connection(self.consistent_hash(group))
         await connection.zrem(key, channel)
@@ -523,7 +521,7 @@ async def group_send(self, group: str, message):
         """
         Sends a message to the entire group.
         """
-        assert self.require_valid_group_name(group), "Group name not valid"
+        assert self.valid_group_name(group), "Group name not valid"
         # Retrieve list of all channel names
         key = self._group_key(group)
         connection = self.connection(self.consistent_hash(group))
@@ -671,7 +669,7 @@ def deserialize(self, message: bytes):
 
     ### Internal functions ###
 
-    def consistent_hash(self, value: "Union[str, Buffer]") -> int:
+    def consistent_hash(self, value: typing.Union[str, "Buffer"]) -> int:
         return _consistent_hash(value, self.ring_size)
 
     def __str__(self):

channels_redis/pubsub.py

@@ -13,15 +13,16 @@
     create_pool,
     decode_hosts,
 )
-from typing import TYPE_CHECKING, Dict, Union, Optional, Any, Iterable
 
-if TYPE_CHECKING:
+import typing
+
+if typing.TYPE_CHECKING:
     from redis.asyncio.client import Redis
     from typing_extensions import Buffer
 logger = logging.getLogger(__name__)
 
 
-async def _async_proxy(obj: "RedisPubSubChannelLayer", name: "str", *args, **kwargs):
+async def _async_proxy(obj: "RedisPubSubChannelLayer", name: str, *args, **kwargs):
     # Must be defined as a function and not a method due to
     # https://bugs.python.org/issue38364
     layer = obj._get_layer()
@@ -32,13 +33,15 @@ class RedisPubSubChannelLayer:
     def __init__(
         self,
         *args,
-        symmetric_encryption_keys: "Optional[Iterable[Union[str, Buffer]]]" = None,
+        symmetric_encryption_keys: typing.Optional[
+            typing.Iterable[typing.Union[str, "Buffer"]]
+        ] = None,
         serializer_format="msgpack",
         **kwargs,
     ):
         self._args = args
         self._kwargs = kwargs
-        self._layers: "Dict[asyncio.AbstractEventLoop, RedisPubSubLoopLayer]" = {}
+        self._layers: typing.Dict[asyncio.AbstractEventLoop, RedisPubSubLoopLayer] = {}
         # serialization
         self._serializer = registry.get_serializer(
             serializer_format,
@@ -95,11 +98,11 @@ class RedisPubSubLoopLayer:
 
     def __init__(
         self,
-        hosts: "Union[Iterable, str, bytes, None]" = None,
-        prefix="asgi",
+        hosts: typing.Union[typing.Iterable, str, bytes, None] = None,
+        prefix: str = "asgi",
         on_disconnect=None,
         on_reconnect=None,
-        channel_layer: "Optional[RedisPubSubChannelLayer]" = None,
+        channel_layer: typing.Optional[RedisPubSubChannelLayer] = None,
         **kwargs,
     ):
         self.prefix = prefix
@@ -110,7 +113,7 @@ def __init__(
 
         # Each consumer gets its own *specific* channel, created with the `new_channel()` method.
         # This dict maps `channel_name` to a queue of messages for that channel.
-        self.channels: "Dict[Any, asyncio.Queue]" = {}
+        self.channels: typing.Dict[typing.Any, asyncio.Queue] = {}
 
         # A channel can subscribe to zero or more groups.
         # This dict maps `group_name` to set of channel names who are subscribed to that group.
@@ -122,7 +125,7 @@ def __init__(
         ]
 
     def _get_shard(
-        self, channel_or_group_name: "Union[str, Buffer]"
+        self, channel_or_group_name: typing.Union[str, "Buffer"]
     ) -> "RedisSingleShardConnection":
         """
         Return the shard that is used exclusively for this channel or group.
@@ -265,17 +268,21 @@ async def flush(self):
 
 
 class RedisSingleShardConnection:
-    def __init__(self, host: "Dict[str, Any]", channel_layer: "RedisPubSubLoopLayer"):
+    def __init__(
+        self, host: typing.Dict[str, typing.Any], channel_layer: RedisPubSubLoopLayer
+    ):
         self.host = host
         self.channel_layer = channel_layer
         self._subscribed_to = set()
         self._lock = asyncio.Lock()
-        self._redis: "Optional[Redis]" = None
+        self._redis: typing.Optional["Redis"] = None
         self._pubsub = None
-        self._receive_task: "Optional[asyncio.Task]" = None
+        self._receive_task: typing.Optional[asyncio.Task] = None
 
     async def publish(
-        self, channel: "Union[str, bytes]", message: "Union[str, bytes, int, float]"
+        self,
+        channel: typing.Union[str, bytes],
+        message: typing.Union[str, bytes, int, float],
     ):
         async with self._lock:
             self._ensure_redis()
@@ -335,7 +342,7 @@ async def _do_receiving(self):
                 logger.exception("Unexpected exception in receive task")
                 await asyncio.sleep(1)
 
-    def _receive_message(self, message: "Optional[Dict]"):
+    def _receive_message(self, message: typing.Optional[typing.Dict]):
         if message is not None:
             name = message["channel"]
             data = message["data"]

channels_redis/serializers.py

@@ -3,9 +3,9 @@
 import hashlib
 import json
 import random
-from typing import TYPE_CHECKING, Dict, Union, Optional, Any, Iterable, Type, cast
+import typing
 
-if TYPE_CHECKING:
+if typing.TYPE_CHECKING:
     from typing_extensions import Buffer
 
 try:
@@ -24,19 +24,23 @@ class SerializerDoesNotExist(KeyError):
 class BaseMessageSerializer(abc.ABC):
     def __init__(
         self,
-        symmetric_encryption_keys: "Optional[Iterable[Union[str, Buffer]]]" = None,
+        symmetric_encryption_keys: typing.Optional[
+            typing.Iterable[typing.Union[str, "Buffer"]]
+        ] = None,
         random_prefix_length: int = 0,
-        expiry: "Optional[int]" = None,
+        expiry: typing.Optional[int] = None,
     ):
         self.random_prefix_length = random_prefix_length
         self.expiry = expiry
-        self.crypter: "Optional[MultiFernet]" = None
+        self.crypter: typing.Optional["MultiFernet"] = None
         # Set up any encryption objects
         self._setup_encryption(symmetric_encryption_keys)
 
     def _setup_encryption(
         self,
-        symmetric_encryption_keys: "Optional[Union[Iterable[Union[str, Buffer]], str, bytes]]",
+        symmetric_encryption_keys: typing.Optional[
+            typing.Union[typing.Iterable[typing.Union[str, "Buffer"]], str, bytes]
+        ],
     ):
         # See if we can do encryption if they asked
         if not symmetric_encryption_keys:
@@ -47,15 +51,17 @@ def _setup_encryption(
             raise ValueError(
                 "symmetric_encryption_keys must be a list of possible keys"
             )
-        keys = cast("Iterable[Union[str, Buffer]]", symmetric_encryption_keys)
+        keys = typing.cast(
+            typing.Iterable[typing.Union[str, "Buffer"]], symmetric_encryption_keys
+        )
         if _MultiFernet is None:
             raise ValueError(
                 "Cannot run with encryption without 'cryptography' installed."
             )
         sub_fernets = [self.make_fernet(key) for key in keys]
         self.crypter = _MultiFernet(sub_fernets)
 
-    def make_fernet(self, key: "Union[str, Buffer]") -> "Fernet":
+    def make_fernet(self, key: typing.Union[str, "Buffer"]) -> "Fernet":
         """
         Given a single encryption key, returns a Fernet instance using it.
         """
@@ -129,7 +135,7 @@ def as_bytes(self, message, *args, **kwargs) -> bytes:
 
 
 # code ready for a future in which msgpack may become an optional dependency
-MsgPackSerializer: "Optional[Type[BaseMessageSerializer]]" = None
+MsgPackSerializer: typing.Optional[typing.Type[BaseMessageSerializer]] = None
 try:
     import msgpack
 except ImportError as exc:
@@ -153,10 +159,10 @@ class SerializersRegistry:
     """
 
     def __init__(self):
-        self._registry: "Dict[Any, Type[BaseMessageSerializer]]" = {}
+        self._registry: typing.Dict[typing.Any, typing.Type[BaseMessageSerializer]] = {}
 
     def register_serializer(
-        self, format, serializer_class: "Type[BaseMessageSerializer]"
+        self, format, serializer_class: typing.Type[BaseMessageSerializer]
     ):
         """
         Register a new serializer for given format
@@ -174,7 +180,7 @@ def register_serializer(
 
         self._registry[format] = serializer_class
 
-    def get_serializer(self, format, *args, **kwargs) -> "BaseMessageSerializer":
+    def get_serializer(self, format, *args, **kwargs) -> BaseMessageSerializer:
         try:
             serializer_class = self._registry[format]
         except KeyError: