Fix type hints

Author: Archmonger

src/js/packages/@reactpy/client/src/mount.tsx

@@ -7,12 +7,17 @@ export function mountReactPy(props: MountProps) {
   // WebSocket route for component rendering
   const wsProtocol = `ws${window.location.protocol === "https:" ? "s" : ""}:`;
   const wsOrigin = `${wsProtocol}//${window.location.host}`;
-  const componentUrl = new URL(`${wsOrigin}${props.pathPrefix}${props.appendComponentPath || ""}`);
+  const componentUrl = new URL(
+    `${wsOrigin}${props.pathPrefix}${props.appendComponentPath || ""}`,
+  );
 
   // Embed the initial HTTP path into the WebSocket URL
   componentUrl.searchParams.append("http_pathname", window.location.pathname);
   if (window.location.search) {
-    componentUrl.searchParams.append("http_query_string", window.location.search);
+    componentUrl.searchParams.append(
+      "http_query_string",
+      window.location.search,
+    );
   }
 
   // Configure a new ReactPy client

src/reactpy/asgi/middleware.py

@@ -5,12 +5,13 @@
 import re
 import traceback
 import urllib.parse
-from collections.abc import Coroutine, Iterable
+from collections.abc import Iterable
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any
 
 import orjson
+from asgiref import typing as asgi_types
 from asgiref.compatibility import guarantee_single_callable
 from servestatic import ServeStaticASGI
 from typing_extensions import Unpack
@@ -33,7 +34,7 @@ class ReactPyMiddleware:
 
     def __init__(
         self,
-        app: Callable[..., Coroutine],
+        app: asgi_types.ASGIApplication,
         root_components: Iterable[str],
         **settings: Unpack[ReactPyConfig],
     ) -> None:
@@ -61,7 +62,7 @@ def __init__(
         self.static_pattern = re.compile(f"^{self.static_path}.*")
 
         # Component attributes
-        self.user_app = guarantee_single_callable(app)
+        self.user_app: asgi_types.ASGI3Application = guarantee_single_callable(app)  # type: ignore
         self.root_components = import_components(root_components)
 
         # Directory attributes
@@ -84,9 +85,9 @@ def __init__(
 
     async def __call__(
         self,
-        scope: dict[str, Any],
-        receive: Callable[..., Coroutine],
-        send: Callable[..., Coroutine],
+        scope: asgi_types.Scope,
+        receive: asgi_types.ASGIReceiveCallable,
+        send: asgi_types.ASGISendCallable,
     ) -> None:
         """The ASGI entrypoint that determines whether ReactPy should route the
         request to ourselves or to the user application."""
@@ -105,13 +106,13 @@ async def __call__(
         # Serve the user's application
         await self.user_app(scope, receive, send)
 
-    def match_dispatch_path(self, scope: dict) -> bool:
+    def match_dispatch_path(self, scope: asgi_types.WebSocketScope) -> bool:
         return bool(re.match(self.dispatcher_pattern, scope["path"]))
 
-    def match_static_path(self, scope: dict) -> bool:
+    def match_static_path(self, scope: asgi_types.HTTPScope) -> bool:
         return bool(re.match(self.static_pattern, scope["path"]))
 
-    def match_web_modules_path(self, scope: dict) -> bool:
+    def match_web_modules_path(self, scope: asgi_types.HTTPScope) -> bool:
         return bool(re.match(self.js_modules_pattern, scope["path"]))
 
 
@@ -121,19 +122,21 @@ class ComponentDispatchApp:
 
     async def __call__(
         self,
-        scope: dict[str, Any],
-        receive: Callable[..., Coroutine],
-        send: Callable[..., Coroutine],
+        scope: asgi_types.WebSocketScope,
+        receive: asgi_types.ASGIReceiveCallable,
+        send: asgi_types.ASGISendCallable,
     ) -> None:
         """ASGI app for rendering ReactPy Python components."""
-        dispatcher: asyncio.Task | None = None
-        recv_queue: asyncio.Queue = asyncio.Queue()
+        dispatcher: asyncio.Task[Any] | None = None
+        recv_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
 
         # Start a loop that handles ASGI websocket events
         while True:
             event = await receive()
             if event["type"] == "websocket.connect":
-                await send({"type": "websocket.accept"})
+                await send(
+                    {"type": "websocket.accept", "subprotocol": None, "headers": []}
+                )
                 dispatcher = asyncio.create_task(
                     self.run_dispatcher(scope, receive, send, recv_queue)
                 )
@@ -143,16 +146,16 @@ async def __call__(
                     dispatcher.cancel()
                 break
 
-            elif event["type"] == "websocket.receive":
+            elif event["type"] == "websocket.receive" and event["text"]:
                 queue_put_func = recv_queue.put(orjson.loads(event["text"]))
                 await queue_put_func
 
     async def run_dispatcher(
         self,
-        scope: dict[str, Any],
-        receive: Callable[..., Coroutine],
-        send: Callable[..., Coroutine],
-        recv_queue: asyncio.Queue,
+        scope: asgi_types.WebSocketScope,
+        receive: asgi_types.ASGIReceiveCallable,
+        send: asgi_types.ASGISendCallable,
+        recv_queue: asyncio.Queue[dict[str, Any]],
     ) -> None:
         """Asyncio background task that renders and transmits layout updates of ReactPy components."""
         try:
@@ -187,11 +190,15 @@ async def run_dispatcher(
 
             # Start the ReactPy component rendering loop
             await serve_layout(
-                Layout(ConnectionContext(component(), value=connection)),  # type: ignore
+                Layout(ConnectionContext(component(), value=connection)),
                 lambda msg: send(
-                    {"type": "websocket.send", "text": orjson.dumps(msg).decode()}
+                    {
+                        "type": "websocket.send",
+                        "text": orjson.dumps(msg).decode(),
+                        "bytes": None,
+                    }
                 ),
-                recv_queue.get,
+                recv_queue.get,  # type: ignore
             )
 
         # Manually log exceptions since this function is running in a separate asyncio task.
@@ -206,9 +213,9 @@ class StaticFileApp:
 
     async def __call__(
         self,
-        scope: dict[str, Any],
-        receive: Callable[..., Coroutine],
-        send: Callable[..., Coroutine],
+        scope: asgi_types.HTTPScope,
+        receive: asgi_types.ASGIReceiveCallable,
+        send: asgi_types.ASGISendCallable,
     ) -> None:
         """ASGI app for ReactPy static files."""
         if not self._static_file_server:
@@ -218,7 +225,7 @@ async def __call__(
                 prefix=self.parent.static_path,
             )
 
-        return await self._static_file_server(scope, receive, send)
+        await self._static_file_server(scope, receive, send)
 
 
 @dataclass
@@ -228,9 +235,9 @@ class WebModuleApp:
 
     async def __call__(
         self,
-        scope: dict[str, Any],
-        receive: Callable[..., Coroutine],
-        send: Callable[..., Coroutine],
+        scope: asgi_types.HTTPScope,
+        receive: asgi_types.ASGIReceiveCallable,
+        send: asgi_types.ASGISendCallable,
     ) -> None:
         """ASGI app for ReactPy web modules."""
         if not self._static_file_server:
@@ -241,4 +248,4 @@ async def __call__(
                 autorefresh=True,
             )
 
-        return await self._static_file_server(scope, receive, send)
+        await self._static_file_server(scope, receive, send)

src/reactpy/asgi/standalone.py

@@ -2,13 +2,12 @@
 
 import hashlib
 import re
-from collections.abc import Coroutine
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from email.utils import formatdate
 from logging import getLogger
-from typing import Any, Callable
 
+from asgiref import typing as asgi_types
 from typing_extensions import Unpack
 
 from reactpy import html
@@ -61,9 +60,9 @@ class ReactPyApp:
 
     async def __call__(
         self,
-        scope: dict[str, Any],
-        receive: Callable[..., Coroutine],
-        send: Callable[..., Coroutine],
+        scope: asgi_types.Scope,
+        receive: asgi_types.ASGIReceiveCallable,
+        send: asgi_types.ASGISendCallable,
     ) -> None:
         if scope["type"] != "http":
             if scope["type"] != "lifespan":
@@ -121,7 +120,7 @@ async def __call__(
             headers=dict_to_byte_list(response_headers),
         )
 
-    def match_dispatch_path(self, scope: dict) -> bool:
+    def match_dispatch_path(self, scope: asgi_types.WebSocketScope) -> bool:
         """Method override to remove `dotted_path` from the dispatcher URL."""
         return str(scope["path"]) == self.parent.dispatcher_path
 

src/reactpy/asgi/utils.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Coroutine, Iterable, Sequence
+from collections.abc import Iterable
 from importlib import import_module
-from typing import Any, Callable
+from typing import Any
+
+from asgiref import typing as asgi_types
 
 from reactpy._option import Option
 from reactpy.types import ReactPyConfig, VdomDict
@@ -69,15 +71,24 @@ def vdom_head_to_html(head: VdomDict) -> str:
 
 async def http_response(
     *,
-    send: Callable[[dict[str, Any]], Coroutine],
+    send: asgi_types.ASGISendCallable,
     method: str,
     code: int = 200,
     message: str = "",
-    headers: Sequence = (),
+    headers: Iterable[tuple[bytes, bytes]] = (),
 ) -> None:
     """Sends a HTTP response using the ASGI `send` API."""
-    start_msg = {"type": "http.response.start", "status": code, "headers": [*headers]}
-    body_msg: dict[str, str | bytes] = {"type": "http.response.body"}
+    start_msg: asgi_types.HTTPResponseStartEvent = {
+        "type": "http.response.start",
+        "status": code,
+        "headers": [*headers],
+        "trailers": False,
+    }
+    body_msg: asgi_types.HTTPResponseBodyEvent = {
+        "type": "http.response.body",
+        "body": b"",
+        "more_body": False,
+    }
 
     # Add the content type and body to everything other than a HEAD request
     if method != "HEAD":
@@ -87,14 +98,14 @@ async def http_response(
     await send(body_msg)
 
 
-def process_settings(settings: ReactPyConfig):
+def process_settings(settings: ReactPyConfig) -> None:
     """Process the settings and return the final configuration."""
     from reactpy import config
 
     for setting in settings:
         config_name = f"REACTPY_{setting.upper()}"
-        config_object: Option | None = getattr(config, config_name, None)
+        config_object: Option[Any] | None = getattr(config, config_name, None)
         if config_object:
-            config_object.set_current(settings[setting])
+            config_object.set_current(settings[setting])  # type: ignore
         else:
             raise ValueError(f"Unknown ReactPy setting {setting!r}.")

src/reactpy/core/hooks.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import contextlib
-from collections.abc import Coroutine, MutableMapping, Sequence
+from collections.abc import Coroutine, Sequence
 from logging import getLogger
 from types import FunctionType
 from typing import (
@@ -16,6 +16,7 @@
     overload,
 )
 
+from asgiref import typing as asgi_types
 from typing_extensions import TypeAlias
 
 from reactpy.config import REACTPY_DEBUG
@@ -262,7 +263,7 @@ def use_connection() -> Connection[Any]:
     return conn
 
 
-def use_scope() -> MutableMapping[str, Any]:
+def use_scope() -> asgi_types.HTTPScope | asgi_types.WebSocketScope:
     """Get the current :class:`~reactpy.types.Connection`'s scope."""
     return use_connection().scope
 

src/reactpy/jinja.py

@@ -6,14 +6,14 @@
 from reactpy.utils import render_mount_template
 
 
-class ReactPyTemplateTag(StandaloneTag):
+class ReactPyTemplateTag(StandaloneTag):  # type: ignore
     """This allows enables a `component` tag to be used in any Jinja2 rendering context,
     as long as this template tag is registered as a Jinja2 extension."""
 
     safe_output = True
     tags: ClassVar[set[str]] = {"component"}
 
-    def render(self, dotted_path: str, **kwargs):
+    def render(self, dotted_path: str, **kwargs: str) -> str:
         return render_mount_template(
             element_id=uuid4().hex,
             class_=kwargs.pop("class", ""),

src/reactpy/testing/backend.py

@@ -2,14 +2,14 @@
 
 import asyncio
 import logging
-from collections.abc import Coroutine
 from contextlib import AsyncExitStack
 from threading import Thread
 from types import TracebackType
 from typing import Any, Callable
 from urllib.parse import urlencode, urlunparse
 
 import uvicorn
+from asgiref import typing as asgi_types
 
 from reactpy.asgi.standalone import ReactPy
 from reactpy.config import REACTPY_TESTS_DEFAULT_TIMEOUT
@@ -43,7 +43,7 @@ class BackendFixture:
 
     def __init__(
         self,
-        app: Callable[..., Coroutine] | None = None,
+        app: asgi_types.ASGIApplication | None = None,
         host: str = "127.0.0.1",
         port: int | None = None,
         timeout: float | None = None,

src/reactpy/types.py

@@ -2,7 +2,7 @@
 
 import sys
 from collections import namedtuple
-from collections.abc import Mapping, MutableMapping, Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from types import TracebackType
@@ -19,6 +19,7 @@
     runtime_checkable,
 )
 
+from asgiref import typing as asgi_types
 from typing_extensions import TypeAlias, TypedDict
 
 CarrierType = TypeVar("CarrierType")
@@ -253,8 +254,8 @@ def value(self) -> _Type:
 class Connection(Generic[CarrierType]):
     """Represents a connection with a client"""
 
-    scope: MutableMapping[str, Any]
-    """An ASGI scope dictionary"""
+    scope: asgi_types.HTTPScope | asgi_types.WebSocketScope
+    """A scope dictionary related to the current connection."""
 
     location: Location
     """The current location (URL)"""