Django channels stubs

pyrightconfig.stricter.json

@@ -31,6 +31,7 @@
         "stubs/boltons",
         "stubs/braintree",
         "stubs/cffi",
+        "stubs/channels",
         "stubs/dateparser",
         "stubs/defusedxml",
         "stubs/docker",

stubs/channels/@tests/django_settings.py

@@ -0,0 +1,12 @@
+SECRET_KEY = "1"
+
+INSTALLED_APPS = (
+    "django.contrib.contenttypes",
+    "django.contrib.sites",
+    "django.contrib.sessions",
+    "django.contrib.messages",
+    "django.contrib.admin.apps.SimpleAdminConfig",
+    "django.contrib.staticfiles",
+    "django.contrib.auth",
+    "channels",
+)

stubs/channels/@tests/stubtest_allowlist.txt

@@ -0,0 +1,15 @@
+# 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
+channels.db.database_sync_to_async
+
+# Set to None on class, but initialized to non-None value in __init__
+channels.generic.websocket.WebsocketConsumer.groups
+channels.generic.websocket.AsyncWebsocketConsumer.groups

stubs/channels/METADATA.toml

@@ -0,0 +1,8 @@
+version = "4.2.*"
+upstream_repository = "https://github.com/django/channels"
+requires = ["django-stubs>=4.2,<5.3", "asgiref"]
+
+[tool.stubtest]
+mypy_plugins = ['mypy_django_plugin.main']
+mypy_plugins_config = {"django-stubs" = {"django_settings_module" = "@tests.django_settings"}}
+stubtest_requirements = ["daphne"]

stubs/channels/channels/__init__.pyi

@@ -0,0 +1,4 @@
+from typing import Final
+
+__version__: Final[str]
+DEFAULT_CHANNEL_LAYER: Final[str]

stubs/channels/channels/apps.pyi

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

stubs/channels/channels/auth.pyi

@@ -0,0 +1,26 @@
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable
+from channels.middleware import BaseMiddleware
+from django.contrib.auth.backends import BaseBackend
+from django.contrib.auth.base_user import AbstractBaseUser
+from django.contrib.auth.models import AnonymousUser
+from django.utils.functional import LazyObject
+
+from .consumer import _ChannelScope
+from .utils import _ChannelApplication
+
+async def get_user(scope: _ChannelScope) -> AbstractBaseUser | AnonymousUser: ...
+async def login(scope: _ChannelScope, user: AbstractBaseUser, backend: BaseBackend | None = None) -> None: ...
+async def logout(scope: _ChannelScope) -> None: ...
+
+# Inherits AbstractBaseUser to improve autocomplete and show this is a lazy proxy for a user.
+# At runtime, it's just a LazyObject that wraps the actual user instance.
+class UserLazyObject(AbstractBaseUser, LazyObject): ...
+
+class AuthMiddleware(BaseMiddleware):
+    def populate_scope(self, scope: _ChannelScope) -> None: ...
+    async def resolve_scope(self, scope: _ChannelScope) -> None: ...
+    async def __call__(
+        self, scope: _ChannelScope, receive: ASGIReceiveCallable, send: ASGISendCallable
+    ) -> _ChannelApplication: ...
+
+def AuthMiddlewareStack(inner: _ChannelApplication) -> _ChannelApplication: ...

stubs/channels/channels/consumer.pyi

@@ -0,0 +1,75 @@
+from collections.abc import Awaitable
+from typing import Any, ClassVar, Protocol, TypedDict, type_check_only
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope, WebSocketScope
+from channels.auth import UserLazyObject
+from channels.db import database_sync_to_async
+from channels.layers import BaseChannelLayer
+from django.contrib.sessions.backends.base import SessionBase
+from django.utils.functional import LazyObject
+
+# _LazySession is a LazyObject that wraps a SessionBase instance.
+# We subclass both for type checking purposes to expose SessionBase attributes,
+# and suppress mypy's "misc" error with `# type: ignore[misc]`.
+@type_check_only
+class _LazySession(SessionBase, LazyObject):  # type: ignore[misc]
+    _wrapped: SessionBase
+
+@type_check_only
+class _URLRoute(TypedDict):
+    # Values extracted from Django's URLPattern matching,
+    # passed through ASGI scope routing.
+    # `args` and `kwargs` are the result of pattern matching against the URL path.
+    args: tuple[Any, ...]
+    kwargs: dict[str, Any]
+
+# Channel Scope definition
+@type_check_only
+class _ChannelScope(WebSocketScope, total=False):
+    # Channels specific
+    channel: str
+    url_route: _URLRoute
+    path_remaining: str
+
+    # Auth specific
+    cookies: dict[str, str]
+    session: _LazySession
+    user: UserLazyObject | None
+
+# Accepts any ASGI message dict with a required "type" key (str),
+# but allows additional arbitrary keys for flexibility.
+def get_handler_name(message: dict[str, Any]) -> str: ...
+@type_check_only
+class _ASGIApplicationProtocol(Protocol):
+    consumer_class: AsyncConsumer
+
+    # Accepts any initialization kwargs passed to the consumer class.
+    # Typed as `Any` to allow flexibility in subclass-specific arguments.
+    consumer_initkwargs: Any
+
+    def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> Awaitable[None]: ...
+
+class AsyncConsumer:
+    channel_layer_alias: ClassVar[str]
+
+    scope: _ChannelScope
+    channel_layer: BaseChannelLayer
+    channel_name: str
+    channel_receive: ASGIReceiveCallable
+    base_send: ASGISendCallable
+
+    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: ...
+
+class SyncConsumer(AsyncConsumer):
+
+    # Since we're overriding asynchronous methods with synchronous ones,
+    # we need to use `# type: ignore[override]` to suppress mypy errors.
+    @database_sync_to_async
+    def dispatch(self, message: dict[str, Any]) -> None: ...  # type: ignore[override]
+    def send(self, message: dict[str, Any]) -> None: ...  # type: ignore[override]

stubs/channels/channels/db.pyi

@@ -0,0 +1,32 @@
+import asyncio
+from _typeshed import OptExcInfo
+from asyncio import BaseEventLoop
+from collections.abc import Callable, Coroutine
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any, TypeVar
+from typing_extensions import ParamSpec
+
+from asgiref.sync import SyncToAsync
+
+_P = ParamSpec("_P")
+_R = TypeVar("_R")
+
+class DatabaseSyncToAsync(SyncToAsync[_P, _R]):
+    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
+# higher-order function behavior with correct type hints.
+# A direct assignment would result in incorrect type inference for the wrapped function.
+def database_sync_to_async(
+    func: Callable[_P, _R], thread_sensitive: bool = True, executor: ThreadPoolExecutor | None = None
+) -> Callable[_P, Coroutine[Any, Any, _R]]: ...
+async def aclose_old_connections() -> None: ...

stubs/channels/channels/exceptions.pyi

@@ -0,0 +1,8 @@
+class RequestAborted(Exception): ...
+class RequestTimeout(RequestAborted): ...
+class InvalidChannelLayerError(ValueError): ...
+class AcceptConnection(Exception): ...
+class DenyConnection(Exception): ...
+class ChannelFull(Exception): ...
+class MessageTooLarge(Exception): ...
+class StopConsumer(Exception): ...

stubs/channels/channels/generic/http.pyi

@@ -0,0 +1,19 @@
+from _typeshed import Unused
+from collections.abc import Iterable
+from typing import Any
+
+from asgiref.typing import HTTPDisconnectEvent, HTTPRequestEvent, HTTPScope
+from channels.consumer import AsyncConsumer
+
+class AsyncHttpConsumer(AsyncConsumer):
+    body: list[bytes]
+    scope: HTTPScope  # type: ignore[assignment]
+
+    def __init__(self, *args: Unused, **kwargs: Unused) -> None: ...
+    async def send_headers(self, *, status: int = 200, headers: Iterable[tuple[bytes, bytes]] | None = None) -> None: ...
+    async def send_body(self, body: bytes, *, more_body: bool = False) -> None: ...
+    async def send_response(self, status: int, body: bytes, **kwargs: Any) -> None: ...
+    async def handle(self, body: bytes) -> None: ...
+    async def disconnect(self) -> None: ...
+    async def http_request(self, message: HTTPRequestEvent) -> None: ...
+    async def http_disconnect(self, message: HTTPDisconnectEvent) -> None: ...

stubs/channels/channels/generic/websocket.pyi

@@ -0,0 +1,61 @@
+from _typeshed import Unused
+from typing import Any
+
+from asgiref.typing import WebSocketConnectEvent, WebSocketDisconnectEvent, WebSocketReceiveEvent
+from channels.consumer import AsyncConsumer, SyncConsumer
+
+class WebsocketConsumer(SyncConsumer):
+    groups: list[str]
+
+    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: ...
+    def websocket_receive(self, message: WebSocketReceiveEvent) -> None: ...
+    def receive(self, text_data: str | None = None, bytes_data: bytes | None = None) -> None: ...
+    def send(  # type: ignore[override]
+        self, text_data: str | None = None, bytes_data: bytes | None = None, close: bool = False
+    ) -> None: ...
+    def close(self, code: int | bool | None = None, reason: str | None = None) -> None: ...
+    def websocket_disconnect(self, message: WebSocketDisconnectEvent) -> None: ...
+    def disconnect(self, code: int) -> None: ...
+
+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: ...  # Returns Any like json.loads()
+    @classmethod
+    def encode_json(cls, content: Any) -> str: ...  # Accepts Any like json.dumps()
+
+class AsyncWebsocketConsumer(AsyncConsumer):
+    groups: list[str]
+
+    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: ...
+    async def websocket_receive(self, message: WebSocketReceiveEvent) -> None: ...
+    async def receive(self, text_data: str | None = None, bytes_data: bytes | None = None) -> None: ...
+    async def send(  # type: ignore[override]
+        self, text_data: str | None = None, bytes_data: bytes | None = None, close: bool = False
+    ) -> None: ...
+    async def close(self, code: int | bool | None = None, reason: str | None = None) -> None: ...
+    async def websocket_disconnect(self, message: WebSocketDisconnectEvent) -> None: ...
+    async def disconnect(self, code: int) -> None: ...
+
+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: ...  # Returns Any like json.loads()
+    @classmethod
+    async def encode_json(cls, content: Any) -> str: ...  # Accepts Any like json.dumps()

stubs/channels/channels/layers.pyi

@@ -0,0 +1,91 @@
+import asyncio
+from re import Pattern
+from typing import Any, ClassVar, overload
+from typing_extensions import TypeAlias, deprecated
+
+class ChannelLayerManager:
+    backends: dict[str, BaseChannelLayer]
+
+    def __init__(self) -> None: ...
+    @property
+    def configs(self) -> dict[str, Any]: ...
+    def make_backend(self, name: str) -> BaseChannelLayer: ...
+    def make_test_backend(self, name: str) -> Any: ...
+    def __getitem__(self, key: str) -> BaseChannelLayer: ...
+    def __contains__(self, key: str) -> bool: ...
+    def set(self, key: str, layer: BaseChannelLayer) -> BaseChannelLayer | None: ...
+
+_ChannelCapacityPattern: TypeAlias = Pattern[str] | str
+_ChannelCapacityDict: TypeAlias = dict[_ChannelCapacityPattern, int]
+_CompiledChannelCapacity: TypeAlias = list[tuple[Pattern[str], int]]
+
+class BaseChannelLayer:
+    MAX_NAME_LENGTH: ClassVar[int] = 100
+    expiry: int
+    capacity: int
+    channel_capacity: _ChannelCapacityDict
+    channel_name_regex: Pattern[str]
+    group_name_regex: Pattern[str]
+    invalid_name_error: str
+
+    def __init__(self, expiry: int = 60, capacity: int = 100, channel_capacity: _ChannelCapacityDict | None = None) -> None: ...
+    def compile_capacities(self, channel_capacity: _ChannelCapacityDict) -> _CompiledChannelCapacity: ...
+    def get_capacity(self, channel: str) -> int: ...
+    @overload
+    def match_type_and_length(self, name: str) -> bool: ...
+    @overload
+    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: object, receive: bool = False) -> bool: ...
+    @overload
+    def require_valid_group_name(self, name: str) -> bool: ...
+    @overload
+    def require_valid_group_name(self, name: object) -> bool: ...
+    @overload
+    def valid_channel_names(self, names: list[str], receive: bool = False) -> bool: ...
+    @overload
+    def valid_channel_names(self, names: list[Any], receive: bool = False) -> bool: ...
+    def non_local_name(self, name: str) -> str: ...
+    async def send(self, channel: str, message: dict[str, Any]) -> None: ...
+    async def receive(self, channel: str) -> dict[str, Any]: ...
+    async def new_channel(self) -> str: ...
+    async def flush(self) -> None: ...
+    async def group_add(self, group: str, channel: str) -> None: ...
+    async def group_discard(self, group: str, channel: str) -> None: ...
+    async def group_send(self, group: str, message: dict[str, Any]) -> None: ...
+    @deprecated("Use require_valid_channel_name instead.")
+    def valid_channel_name(self, channel_name: str, receive: bool = False) -> bool: ...
+    @deprecated("Use require_valid_group_name instead.")
+    def valid_group_name(self, group_name: str) -> bool: ...
+
+_InMemoryQueueData: TypeAlias = tuple[float, dict[str, Any]]
+
+class InMemoryChannelLayer(BaseChannelLayer):
+    channels: dict[str, asyncio.Queue[_InMemoryQueueData]]
+    groups: dict[str, dict[str, float]]
+    group_expiry: int
+
+    def __init__(
+        self,
+        expiry: int = 60,
+        group_expiry: int = 86400,
+        capacity: int = 100,
+        channel_capacity: _ChannelCapacityDict | None = ...,
+    ) -> None: ...
+
+    extensions: list[str]
+
+    async def send(self, channel: str, message: dict[str, Any]) -> None: ...
+    async def receive(self, channel: str) -> dict[str, Any]: ...
+    async def new_channel(self, prefix: str = "specific.") -> str: ...
+    async def flush(self) -> None: ...
+    async def close(self) -> None: ...
+    async def group_add(self, group: str, channel: str) -> None: ...
+    async def group_discard(self, group: str, channel: str) -> None: ...
+    async def group_send(self, group: str, message: dict[str, Any]) -> None: ...
+
+def get_channel_layer(alias: str = ...) -> BaseChannelLayer | None: ...
+
+channel_layers: ChannelLayerManager