FIX

Author: iwakitakuma33

channels_redis/core.py

@@ -9,8 +9,8 @@
 
 from redis import asyncio as aioredis
 
-from channels.exceptions import ChannelFull
-from channels.layers import BaseChannelLayer
+from channels.exceptions import ChannelFull  # type: ignore[import-untyped]
+from channels.layers import BaseChannelLayer  # type: ignore[import-untyped]
 
 from .serializers import registry
 from .utils import (
@@ -38,12 +38,10 @@ class ChannelLock:
     """
 
     def __init__(self) -> None:
-        self.locks: collections.defaultdict[str, asyncio.Lock] = (
-            collections.defaultdict(asyncio.Lock)
-        )
-        self.wait_counts: collections.defaultdict[str, int] = collections.defaultdict(
-            int
+        self.locks: typing.DefaultDict[str, asyncio.Lock] = collections.defaultdict(
+            asyncio.Lock
         )
+        self.wait_counts: typing.DefaultDict[str, int] = collections.defaultdict(int)
 
     async def acquire(self, channel: str) -> bool:
         """
@@ -69,7 +67,7 @@ def release(self, channel: str) -> None:
             del self.wait_counts[channel]
 
 
-class BoundedQueue(asyncio.Queue[typing.Any]):
+class BoundedQueue(asyncio.Queue):
     def put_nowait(self, item: typing.Any) -> None:
         if self.full():
             # see: https://github.com/django/channels_redis/issues/212
@@ -157,11 +155,11 @@ def __init__(
         # Event loop they are trying to receive on
         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: typing.DefaultDict[str, BoundedQueue] = (
             collections.defaultdict(functools.partial(BoundedQueue, self.capacity))
         )
         # Detached channel cleanup tasks
-        self.receive_cleaners: typing.List[asyncio.Task[typing.Any]] = []
+        self.receive_cleaners: typing.List["asyncio.Task[typing.Any]"] = []
         # 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()
@@ -240,7 +238,7 @@ async def _brpop_with_clean(
         connection = self.connection(index)
         # Cancellation here doesn't matter, we're not doing anything destructive
         # and the script executes atomically...
-        await connection.eval(cleanup_script, 0, channel, backup_queue)
+        await connection.eval(cleanup_script, 0, channel, backup_queue)  # type: ignore[misc]
         # ...and it doesn't matter here either, the message will be safe in the backup.
         result = await connection.bzpopmin(channel, timeout=timeout)
 
@@ -271,9 +269,9 @@ async def receive(self, channel: str) -> typing.Any:
         assert self.valid_channel_name(channel)
         if "!" in channel:
             real_channel = self.non_local_name(channel)
-            assert real_channel.endswith(self.client_prefix + "!"), (
-                "Wrong client prefix"
-            )
+            assert real_channel.endswith(
+                self.client_prefix + "!"
+            ), "Wrong client prefix"
             # Enter receiving section
             loop = asyncio.get_running_loop()
             self.receive_count += 1
@@ -293,7 +291,7 @@ async def receive(self, channel: str) -> typing.Any:
                 message = None
                 while self.receive_buffer[channel].empty():
                     _tasks = [
-                        self.receive_lock.acquire(),
+                        self.receive_lock.acquire(),  # type: ignore[union-attr]
                         self.receive_buffer[channel].get(),
                     ]
                     tasks = [asyncio.ensure_future(task) for task in _tasks]
@@ -312,7 +310,7 @@ async def receive(self, channel: str) -> typing.Any:
                             if not task.cancel():
                                 assert task.done()
                                 if task.result() is True:
-                                    self.receive_lock.release()
+                                    self.receive_lock.release()  # type: ignore[union-attr]
 
                         raise
 
@@ -335,7 +333,7 @@ async def receive(self, channel: str) -> typing.Any:
                     if message or exception:
                         if token:
                             # We will not be receving as we already have the message.
-                            self.receive_lock.release()
+                            self.receive_lock.release()  # type: ignore[union-attr]
 
                         if exception:
                             raise exception
@@ -362,7 +360,7 @@ async def receive(self, channel: str) -> typing.Any:
                             del self.receive_buffer[channel]
                             raise
                         finally:
-                            self.receive_lock.release()
+                            self.receive_lock.release()  # type: ignore[union-attr]
 
                 # We know there's a message available, because there
                 # couldn't have been any interruption between empty() and here
@@ -377,7 +375,7 @@ async def receive(self, channel: str) -> typing.Any:
                 self.receive_count -= 1
                 # If we were the last out, drop the receive lock
                 if self.receive_count == 0:
-                    assert not self.receive_lock.locked()
+                    assert not self.receive_lock.locked()  # type: ignore[union-attr]
                     self.receive_lock = None
                     self.receive_event_loop = None
         else:
@@ -422,7 +420,7 @@ async def receive_single(
             )
             self.receive_cleaners.append(cleaner)
 
-            def _cleanup_done(cleaner: asyncio.Task[typing.Any]) -> None:
+            def _cleanup_done(cleaner: "asyncio.Task") -> None:
                 self.receive_cleaners.remove(cleaner)
                 self.receive_clean_locks.release(channel_key)
 
@@ -468,7 +466,7 @@ async def flush(self) -> None:
         # Go through each connection and remove all with prefix
         for i in range(self.ring_size):
             connection = self.connection(i)
-            await connection.eval(delete_prefix, 0, self.prefix + "*")
+            await connection.eval(delete_prefix, 0, self.prefix + "*")  # type: ignore[union-attr,misc]
         # Now clear the pools as well
         await self.close_pools()
 
@@ -584,7 +582,7 @@ async def group_send(self, group: str, message: typing.Any) -> None:
 
             # channel_keys does not contain a single redis key more than once
             connection = self.connection(connection_index)
-            channels_over_capacity = await connection.eval(
+            channels_over_capacity = await connection.eval(  # type: ignore[misc]
                 group_send_lua, len(channel_redis_keys), *channel_redis_keys, *args
             )
             _channels_over_capacity = -1.0

channels_redis/pubsub.py

@@ -1,4 +1,5 @@
 import asyncio
+from copy import deepcopy
 import functools
 import logging
 import typing
@@ -16,7 +17,7 @@
 )
 
 if typing.TYPE_CHECKING:
-    from redis.asyncio.client import Redis, PubSub
+    from redis.asyncio.client import PubSub, Redis
     from typing_extensions import Buffer
 logger = logging.getLogger(__name__)
 
@@ -81,10 +82,11 @@ def _get_layer(self) -> "RedisPubSubLoopLayer":
         try:
             layer = self._layers[loop]
         except KeyError:
+            kwargs = deepcopy(self._kwargs)
+            kwargs["channel_layer"] = self
             layer = RedisPubSubLoopLayer(
                 *self._args,
-                **self._kwargs,
-                channel_layer=self,
+                **kwargs,
             )
             self._layers[loop] = layer
             _wrap_close(self, loop)
@@ -163,7 +165,7 @@ async def send(
         Send a message onto a (general or specific) channel.
         """
         shard = self._get_shard(channel)
-        await shard.publish(channel, self.channel_layer.serialize(message))
+        await shard.publish(channel, self.channel_layer.serialize(message))  # type: ignore[union-attr]
 
     async def new_channel(self, prefix: str = "specific.") -> str:
         """
@@ -205,7 +207,7 @@ async def receive(self, channel: typing.Union[str, bytes]) -> typing.Any:
                     # We don't re-raise here because we want the CancelledError to be the one re-raised.
             raise
 
-        return self.channel_layer.deserialize(message)
+        return self.channel_layer.deserialize(message)  # type: ignore[union-attr]
 
     ################################################################################
     # Groups extension
@@ -254,7 +256,7 @@ async def group_send(self, group: str, message: typing.Any) -> None:
         """
         group_channel = self._get_group_channel_name(group)
         shard = self._get_shard(group_channel)
-        await shard.publish(group_channel, self.channel_layer.serialize(message))
+        await shard.publish(group_channel, self.channel_layer.serialize(message))  # type: ignore[union-attr]
 
     ################################################################################
     # Flush extension
@@ -282,7 +284,7 @@ def __init__(
         self._lock = asyncio.Lock()
         self._redis: typing.Optional["Redis"] = None
         self._pubsub: typing.Optional["PubSub"] = None
-        self._receive_task: typing.Optional[asyncio.Task[typing.Any]] = None
+        self._receive_task: typing.Optional["asyncio.Task[typing.Any]"] = None
 
     async def publish(
         self,
@@ -291,22 +293,22 @@ async def publish(
     ) -> None:
         async with self._lock:
             self._ensure_redis()
-            await self._redis.publish(channel, message)
+            await self._redis.publish(channel, message)  # type: ignore[union-attr]
 
     async def subscribe(self, channel: typing.Union[str, bytes]) -> None:
         async with self._lock:
             if channel not in self._subscribed_to:
                 self._ensure_redis()
                 self._ensure_receiver()
-                await self._pubsub.subscribe(channel)
+                await self._pubsub.subscribe(channel)  # type: ignore[union-attr]
                 self._subscribed_to.add(channel)
 
     async def unsubscribe(self, channel: typing.Union[str, bytes]) -> None:
         async with self._lock:
             if channel in self._subscribed_to:
                 self._ensure_redis()
                 self._ensure_receiver()
-                await self._pubsub.unsubscribe(channel)
+                await self._pubsub.unsubscribe(channel)  # type: ignore[union-attr]
                 self._subscribed_to.remove(channel)
 
     async def flush(self) -> None:

channels_redis/serializers.py

@@ -128,15 +128,15 @@ def as_bytes(self, message: typing.Any, *args, **kwargs) -> bytes:
         msg = json.dumps(message, *args, **kwargs)
         return msg.encode("utf-8")
 
-    from_bytes = staticmethod(json.loads)
+    from_bytes = staticmethod(json.loads)  # type: ignore[assignment]
 
 
 # code ready for a future in which msgpack may become an optional dependency
 MsgPackSerializer: typing.Union[
     typing.Type[BaseMessageSerializer], typing.Type[MissingSerializer]
 ]
 try:
-    import msgpack
+    import msgpack  # type: ignore[import-untyped]
 except ImportError as exc:
 
     class _MsgPackSerializer(MissingSerializer):
@@ -146,8 +146,8 @@ class _MsgPackSerializer(MissingSerializer):
 else:
 
     class __MsgPackSerializer(BaseMessageSerializer):
-        as_bytes = staticmethod(msgpack.packb)
-        from_bytes = staticmethod(msgpack.unpackb)
+        as_bytes = staticmethod(msgpack.packb)  # type: ignore[assignment]
+        from_bytes = staticmethod(msgpack.unpackb)  # type: ignore[assignment]
 
     MsgPackSerializer = __MsgPackSerializer
 

channels_redis/utils.py

@@ -47,7 +47,7 @@ def _wrapper(self, *args, **kwargs) -> typing.Any:
         self.close = original_impl
         return self.close(*args, **kwargs)
 
-    loop.close = types.MethodType(_wrapper, loop)
+    loop.close = types.MethodType(_wrapper, loop)  # type: ignore[method-assign]
 
 
 async def _close_redis(connection: "Redis") -> None:

tox.ini

@@ -7,10 +7,12 @@ envlist =
 [testenv]
 usedevelop = true
 extras = tests
+allowlist_externals = mypy
 commands =
     pytest -v {posargs}
-    mypy . {posargs}
+    ; mypy  --show-error-codes --config-file=tox.ini channels_redis
 deps =
+    mypy>=1.9.0
     ch30: channels>=3.0,<3.1
     ch40: channels>=4.0,<4.1
     chmain: https://github.com/django/channels/archive/main.tar.gz
@@ -28,3 +30,7 @@ commands =
     flake8 channels_redis tests
     black --check channels_redis tests
     isort --check-only --diff channels_redis tests
+
+
+[mypy]
+disallow_untyped_defs = False