Enhance Django Channels type stubs with better annotations and parameter defaults

Author: huynguyengl99

stubs/channels/@tests/stubtest_allowlist.txt

@@ -1,7 +1,10 @@
-# channels.auth.UserLazyObject.DoesNotExist is not present at runtime
-# channels.auth.UserLazyObject.MultipleObjectsReturned is not present at runtime
-# channels.auth.UserLazyObject@AnnotatedWith is not present at runtime
-channels.auth.UserLazyObject.*
+# channels.auth.UserLazyObject metaclass is mismatch
+channels.auth.UserLazyObject
+
+# these one need to be exclude due to mypy error: * is not present at runtime
+channels.auth.UserLazyObject.DoesNotExist
+channels.auth.UserLazyObject.MultipleObjectsReturned
+channels.auth.UserLazyObject@AnnotatedWith
 
 # database_sync_to_async is implemented as a class instance but stubbed as a function
 # for better type inference when used as decorator/function

stubs/channels/channels/apps.pyi

@@ -1,5 +1,7 @@
+from typing import Final
+
 from django.apps import AppConfig
 
 class ChannelsConfig(AppConfig):
-    name: str = "channels"
+    name: Final = "channels"
     verbose_name: str = "Channels"

stubs/channels/channels/consumer.pyi

@@ -61,6 +61,8 @@ class AsyncConsumer:
     async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None: ...
     async def dispatch(self, message: dict[str, Any]) -> None: ...
     async def send(self, message: dict[str, Any]) -> None: ...
+
+    # initkwargs will be used to instantiate the consumer instance.
     @classmethod
     def as_asgi(cls, **initkwargs: Any) -> _ASGIApplicationProtocol: ...
 

stubs/channels/channels/db.pyi

@@ -1,3 +1,5 @@
+import asyncio
+from _typeshed import OptExcInfo
 from asyncio import BaseEventLoop
 from collections.abc import Callable, Coroutine
 from concurrent.futures import ThreadPoolExecutor
@@ -10,7 +12,15 @@ _P = ParamSpec("_P")
 _R = TypeVar("_R")
 
 class DatabaseSyncToAsync(SyncToAsync[_P, _R]):
-    def thread_handler(self, loop: BaseEventLoop, *args: Any, **kwargs: Any) -> Any: ...
+    def thread_handler(
+        self,
+        loop: BaseEventLoop,
+        exc_info: OptExcInfo,
+        task_context: list[asyncio.Task[Any]] | None,
+        func: Callable[_P, _R],
+        *args: _P.args,
+        **kwargs: _P.kwargs,
+    ) -> _R: ...
 
 # We define `database_sync_to_async` as a function instead of assigning
 # `DatabaseSyncToAsync(...)` directly, to preserve both decorator and

stubs/channels/channels/generic/websocket.pyi

@@ -1,18 +1,13 @@
+from _typeshed import Unused
 from typing import Any
 
 from asgiref.typing import WebSocketConnectEvent, WebSocketDisconnectEvent, WebSocketReceiveEvent
-from channels.consumer import AsyncConsumer, SyncConsumer, _ChannelScope
-from channels.layers import BaseChannelLayer
+from channels.consumer import AsyncConsumer, SyncConsumer
 
 class WebsocketConsumer(SyncConsumer):
     groups: list[str] | None
-    scope: _ChannelScope
-    channel_name: str
-    channel_layer: BaseChannelLayer
-    channel_receive: Any
-    base_send: Any
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    def __init__(self, *args: Unused, **kwargs: Unused) -> None: ...
     def websocket_connect(self, message: WebSocketConnectEvent) -> None: ...
     def connect(self) -> None: ...
     def accept(self, subprotocol: str | None = None, headers: list[tuple[str, str]] | None = None) -> None: ...
@@ -27,22 +22,20 @@ class WebsocketConsumer(SyncConsumer):
 
 class JsonWebsocketConsumer(WebsocketConsumer):
     def receive(self, text_data: str | None = None, bytes_data: bytes | None = None, **kwargs: Any) -> None: ...
+    # content is typed as Any to match json.loads() return type - JSON can represent
+    # various Python types (dict, list, str, int, float, bool, None)
     def receive_json(self, content: Any, **kwargs: Any) -> None: ...
+    # content is typed as Any to match json.dumps() input type - accepts any JSON-serializable object
     def send_json(self, content: Any, close: bool = False) -> None: ...
     @classmethod
-    def decode_json(cls, text_data: str) -> Any: ...
+    def decode_json(cls, text_data: str) -> Any: ...  # Returns Any like json.loads()
     @classmethod
-    def encode_json(cls, content: Any) -> str: ...
+    def encode_json(cls, content: Any) -> str: ...  # Accepts Any like json.dumps()
 
 class AsyncWebsocketConsumer(AsyncConsumer):
     groups: list[str] | None
-    scope: _ChannelScope
-    channel_name: str
-    channel_layer: BaseChannelLayer
-    channel_receive: Any
-    base_send: Any
 
-    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    def __init__(self, *args: Unused, **kwargs: Unused) -> None: ...
     async def websocket_connect(self, message: WebSocketConnectEvent) -> None: ...
     async def connect(self) -> None: ...
     async def accept(self, subprotocol: str | None = None, headers: list[tuple[str, str]] | None = None) -> None: ...
@@ -57,9 +50,12 @@ class AsyncWebsocketConsumer(AsyncConsumer):
 
 class AsyncJsonWebsocketConsumer(AsyncWebsocketConsumer):
     async def receive(self, text_data: str | None = None, bytes_data: bytes | None = None, **kwargs: Any) -> None: ...
+    # content is typed as Any to match json.loads() return type - JSON can represent
+    # various Python types (dict, list, str, int, float, bool, None)
     async def receive_json(self, content: Any, **kwargs: Any) -> None: ...
+    # content is typed as Any to match json.dumps() input type - accepts any JSON-serializable object
     async def send_json(self, content: Any, close: bool = False) -> None: ...
     @classmethod
-    async def decode_json(cls, text_data: str) -> Any: ...
+    async def decode_json(cls, text_data: str) -> Any: ...  # Returns Any like json.loads()
     @classmethod
-    async def encode_json(cls, content: Any) -> str: ...
+    async def encode_json(cls, content: Any) -> str: ...  # Accepts Any like json.dumps()

stubs/channels/channels/layers.pyi

@@ -34,15 +34,15 @@ class BaseChannelLayer:
     @overload
     def match_type_and_length(self, name: str) -> bool: ...
     @overload
-    def match_type_and_length(self, name: Any) -> bool: ...
+    def match_type_and_length(self, name: object) -> bool: ...
     @overload
     def require_valid_channel_name(self, name: str, receive: bool = False) -> bool: ...
     @overload
-    def require_valid_channel_name(self, name: Any, receive: bool = False) -> bool: ...
+    def require_valid_channel_name(self, name: object, receive: bool = False) -> bool: ...
     @overload
     def require_valid_group_name(self, name: str) -> bool: ...
     @overload
-    def require_valid_group_name(self, name: Any) -> bool: ...
+    def require_valid_group_name(self, name: object) -> bool: ...
     @overload
     def valid_channel_names(self, names: list[str], receive: bool = False) -> bool: ...
     @overload
@@ -73,7 +73,6 @@ class InMemoryChannelLayer(BaseChannelLayer):
         group_expiry: int = 86400,
         capacity: int = 100,
         channel_capacity: _ChannelCapacityDict | None = ...,
-        **kwargs: Any,
     ) -> None: ...
 
     extensions: list[str]

stubs/channels/channels/management/commands/runworker.pyi

@@ -1,6 +1,7 @@
 import logging
+from _typeshed import Unused
 from argparse import ArgumentParser
-from typing import Any, TypedDict, type_check_only
+from typing import TypedDict, type_check_only
 
 from channels.layers import BaseChannelLayer
 from channels.worker import Worker
@@ -21,4 +22,4 @@ class Command(BaseCommand):
     channel_layer: BaseChannelLayer
 
     def add_arguments(self, parser: ArgumentParser) -> None: ...
-    def handle(self, *args: Any, **options: _RunWorkerCommandOption) -> None: ...
+    def handle(self, *args: Unused, **options: _RunWorkerCommandOption) -> None: ...

stubs/channels/channels/sessions.pyi

@@ -11,6 +11,8 @@ class CookieMiddleware:
     inner: _ChannelApplication
 
     def __init__(self, inner: _ChannelApplication) -> None: ...
+
+    # Returns the same type as the provided _ChannelApplication.
     async def __call__(self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> Any: ...
     @classmethod
     def set_cookie(

stubs/channels/channels/testing/application.pyi

@@ -5,8 +5,13 @@ from asgiref.testing import ApplicationCommunicator as BaseApplicationCommunicat
 def no_op() -> None: ...
 
 class ApplicationCommunicator(BaseApplicationCommunicator):
-    async def send_input(self, message: Any) -> None: ...
-    async def receive_output(self, timeout: float = 1) -> Any: ...
+    # ASGI messages are dictionaries with a "type" key and protocol-specific fields.
+    # Dictionary values can be strings, bytes, lists, or other types depending on the protocol:
+    # - HTTP: {"type": "http.request", "body": b"request data", "headers": [...], ...}
+    # - WebSocket: {"type": "websocket.receive", "bytes": b"binary data"} or {"text": "string"}
+    # - Custom protocols: Application-specific message dictionaries
+    async def send_input(self, message: dict[str, Any]) -> None: ...
+    async def receive_output(self, timeout: float = 1) -> dict[str, Any]: ...
 
     # The following methods are not present in the original source code,
     # but are commonly used in practice. Since the base package doesn't

stubs/channels/channels/utils.pyi

@@ -4,7 +4,7 @@ from typing_extensions import TypeAlias
 
 from asgiref.typing import ASGIApplication, ASGIReceiveCallable
 
-def name_that_thing(thing: Any) -> str: ...
+def name_that_thing(thing: object) -> str: ...
 async def await_many_dispatch(
     consumer_callables: list[Callable[[], Awaitable[ASGIReceiveCallable]]], dispatch: Callable[[dict[str, Any]], Awaitable[None]]
 ) -> None: ...