Django channels stubs

Author: huynguyengl99

pyrightconfig.stricter.json

@@ -32,6 +32,7 @@
         "stubs/braintree",
         "stubs/caldav",
         "stubs/cffi",
+        "stubs/channels",
         "stubs/click-web",
         "stubs/corus",
         "stubs/dateparser",

stubs/channels/METADATA.toml

@@ -0,0 +1,8 @@
+version = "4.*"
+upstream_repository = "https://github.com/django/channels"
+requires = ["django-stubs", "asgiref"]
+requires_python = ">=3.10"
+
+[tool.stubtest]
+skip = true  # due to the need of django mypy plugin config, it should be skipped.
+stubtest_requirements = ["daphne"]

stubs/channels/channels/__init__.pyi

@@ -0,0 +1,2 @@
+__version__: str
+DEFAULT_CHANNEL_LAYER: str

stubs/channels/channels/apps.pyi

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+from django.utils.functional import _StrOrPromise
+
+class ChannelsConfig(AppConfig):
+    name: str = ...
+    verbose_name: _StrOrPromise = ...

stubs/channels/channels/auth.pyi

@@ -0,0 +1,32 @@
+from typing import Any
+
+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, _LazySession
+from .db import database_sync_to_async
+from .utils import _ChannelApplication
+
+@database_sync_to_async
+def get_user(scope: _ChannelScope) -> AbstractBaseUser | AnonymousUser: ...
+@database_sync_to_async
+def login(scope: _ChannelScope, user: AbstractBaseUser, backend: BaseBackend | None = ...) -> None: ...
+@database_sync_to_async
+def logout(scope: _ChannelScope) -> None: ...
+def _get_user_session_key(session: _LazySession) -> Any: ...
+
+class UserLazyObject(AbstractBaseUser, LazyObject):
+    def _setup(self) -> None: ...
+
+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,54 @@
+from collections.abc import Awaitable
+from typing import Any, ClassVar, Protocol
+
+from asgiref.typing import ASGIReceiveCallable, ASGISendCallable, Scope, WebSocketScope
+from channels.auth import UserLazyObject
+from channels.db import database_sync_to_async
+from django.contrib.sessions.backends.base import SessionBase
+from django.utils.functional import LazyObject
+
+class _LazySession(SessionBase, LazyObject):  # type: ignore[misc]
+    _wrapped: SessionBase
+
+# Base ASGI Scope definition
+class _ChannelScope(WebSocketScope, total=False):
+    # Channel specific
+    channel: str
+    url_route: dict[str, Any]
+    path_remaining: str
+
+    # Auth specific
+    cookies: dict[str, str]
+    session: _LazySession
+    user: UserLazyObject | None
+
+def get_handler_name(message: dict[str, Any]) -> str: ...
+
+class _ASGIApplicationProtocol(Protocol):
+    consumer_class: Any
+    consumer_initkwargs: dict[str, Any]
+
+    def __call__(self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> Awaitable[None]: ...
+
+class AsyncConsumer:
+    _sync: ClassVar[bool] = ...
+    channel_layer_alias: ClassVar[str] = ...
+
+    scope: _ChannelScope
+    channel_layer: Any
+    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: ...
+    @classmethod
+    def as_asgi(cls, **initkwargs: Any) -> _ASGIApplicationProtocol: ...
+
+class SyncConsumer(AsyncConsumer):
+    _sync: ClassVar[bool] = ...
+
+    @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,15 @@
+from asyncio import BaseEventLoop
+from collections.abc import Callable, Coroutine
+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, *args: Any, **kwargs: Any) -> Any: ...
+
+def database_sync_to_async(func: Callable[_P, _R]) -> 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/__init__.pyi

@@ -0,0 +1,19 @@
+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: Any, **kwargs: Any) -> None: ...
+    async def send_headers(self, *, status: int = ..., headers: Iterable[tuple[bytes, bytes]] | None = ...) -> None: ...
+    async def send_body(self, body: bytes, *, more_body: bool = ...) -> 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: ...
+    async def send(self, message: dict[str, Any]) -> None: ...  # type: ignore[override]