Overhaul ASGI type hints

Author: Archmonger

src/reactpy/core/hooks.py

@@ -338,9 +338,9 @@ def use_connection() -> Connection[Any]:
     return conn
 
 
-def use_scope() -> asgi_types.HTTPScope | asgi_types.WebSocketScope:
+def use_scope() -> dict[str, Any] | asgi_types.HTTPScope | asgi_types.WebSocketScope:
     """Get the current :class:`~reactpy.types.Connection`'s scope."""
-    return use_connection().scope  # type: ignore
+    return use_connection().scope
 
 
 def use_location() -> Location:

src/reactpy/executors/asgi/middleware.py

@@ -12,7 +12,6 @@
 
 import orjson
 from asgi_tools import ResponseText, ResponseWebSocket
-from asgiref import typing as asgi_types
 from asgiref.compatibility import guarantee_single_callable
 from servestatic import ServeStaticASGI
 from typing_extensions import Unpack
@@ -23,10 +22,18 @@
 from reactpy.core.serve import serve_layout
 from reactpy.executors.asgi.types import (
     AsgiApp,
-    AsgiHttpApp,
-    AsgiLifespanApp,
-    AsgiWebsocketApp,
+    AsgiHttpReceive,
+    AsgiHttpScope,
+    AsgiHttpSend,
+    AsgiReceive,
+    AsgiScope,
+    AsgiSend,
+    AsgiV3App,
+    AsgiV3HttpApp,
+    AsgiV3LifespanApp,
+    AsgiV3WebsocketApp,
     AsgiWebsocketReceive,
+    AsgiWebsocketScope,
     AsgiWebsocketSend,
 )
 from reactpy.executors.utils import check_path, import_components, process_settings
@@ -42,7 +49,7 @@ class ReactPyMiddleware:
 
     def __init__(
         self,
-        app: asgi_types.ASGIApplication,
+        app: AsgiApp,
         root_components: Iterable[str],
         **settings: Unpack[ReactPyConfig],
     ) -> None:
@@ -80,12 +87,12 @@ def __init__(
         )
 
         # User defined ASGI apps
-        self.extra_http_routes: dict[str, AsgiHttpApp] = {}
-        self.extra_ws_routes: dict[str, AsgiWebsocketApp] = {}
-        self.extra_lifespan_app: AsgiLifespanApp | None = None
+        self.extra_http_routes: dict[str, AsgiV3HttpApp] = {}
+        self.extra_ws_routes: dict[str, AsgiV3WebsocketApp] = {}
+        self.extra_lifespan_app: AsgiV3LifespanApp | None = None
 
         # Component attributes
-        self.asgi_app: asgi_types.ASGI3Application = guarantee_single_callable(app)  # type: ignore
+        self.asgi_app: AsgiV3App = guarantee_single_callable(app)  # type: ignore
         self.root_components = import_components(root_components)
 
         # Directory attributes
@@ -98,10 +105,7 @@ def __init__(
         self.web_modules_app = WebModuleApp(parent=self)
 
     async def __call__(
-        self,
-        scope: asgi_types.Scope,
-        receive: asgi_types.ASGIReceiveCallable,
-        send: asgi_types.ASGISendCallable,
+        self, scope: AsgiScope, receive: AsgiReceive, send: AsgiSend
     ) -> None:
         """The ASGI entrypoint that determines whether ReactPy should route the
         request to ourselves or to the user application."""
@@ -125,16 +129,16 @@ async def __call__(
         # Serve the user's application
         await self.asgi_app(scope, receive, send)
 
-    def match_dispatch_path(self, scope: asgi_types.WebSocketScope) -> bool:
+    def match_dispatch_path(self, scope: AsgiWebsocketScope) -> bool:
         return bool(re.match(self.dispatcher_pattern, scope["path"]))
 
-    def match_static_path(self, scope: asgi_types.HTTPScope) -> bool:
+    def match_static_path(self, scope: AsgiHttpScope) -> bool:
         return scope["path"].startswith(self.static_path)
 
-    def match_web_modules_path(self, scope: asgi_types.HTTPScope) -> bool:
+    def match_web_modules_path(self, scope: AsgiHttpScope) -> bool:
         return scope["path"].startswith(self.web_modules_path)
 
-    def match_extra_paths(self, scope: asgi_types.Scope) -> AsgiApp | None:
+    def match_extra_paths(self, scope: AsgiScope) -> AsgiApp | None:
         # Custom defined routes are unused by default to encourage users to handle
         # routing within their ASGI framework of choice.
         return None
@@ -146,13 +150,13 @@ class ComponentDispatchApp:
 
     async def __call__(
         self,
-        scope: asgi_types.WebSocketScope,
-        receive: asgi_types.ASGIReceiveCallable,
-        send: asgi_types.ASGISendCallable,
+        scope: AsgiWebsocketScope,
+        receive: AsgiWebsocketReceive,
+        send: AsgiWebsocketSend,
     ) -> None:
         """ASGI app for rendering ReactPy Python components."""
         # Start a loop that handles ASGI websocket events
-        async with ReactPyWebsocket(scope, receive, send, parent=self.parent) as ws:  # type: ignore
+        async with ReactPyWebsocket(scope, receive, send, parent=self.parent) as ws:
             while True:
                 # Wait for the webserver to notify us of a new event
                 event: dict[str, Any] = await ws.receive(raw=True)  # type: ignore
@@ -175,7 +179,7 @@ async def __call__(
 class ReactPyWebsocket(ResponseWebSocket):
     def __init__(
         self,
-        scope: asgi_types.WebSocketScope,
+        scope: AsgiWebsocketScope,
         receive: AsgiWebsocketReceive,
         send: AsgiWebsocketSend,
         parent: ReactPyMiddleware,
@@ -250,10 +254,7 @@ class StaticFileApp:
     _static_file_server: ServeStaticASGI | None = None
 
     async def __call__(
-        self,
-        scope: asgi_types.HTTPScope,
-        receive: asgi_types.ASGIReceiveCallable,
-        send: asgi_types.ASGISendCallable,
+        self, scope: AsgiHttpScope, receive: AsgiHttpReceive, send: AsgiHttpSend
     ) -> None:
         """ASGI app for ReactPy static files."""
         if not self._static_file_server:
@@ -272,10 +273,7 @@ class WebModuleApp:
     _static_file_server: ServeStaticASGI | None = None
 
     async def __call__(
-        self,
-        scope: asgi_types.HTTPScope,
-        receive: asgi_types.ASGIReceiveCallable,
-        send: asgi_types.ASGISendCallable,
+        self, scope: AsgiHttpScope, receive: AsgiHttpReceive, send: AsgiHttpSend
     ) -> None:
         """ASGI app for ReactPy web modules."""
         if not self._static_file_server:
@@ -291,10 +289,7 @@ async def __call__(
 
 class Error404App:
     async def __call__(
-        self,
-        scope: asgi_types.HTTPScope,
-        receive: asgi_types.ASGIReceiveCallable,
-        send: asgi_types.ASGISendCallable,
+        self, scope: AsgiScope, receive: AsgiReceive, send: AsgiSend
     ) -> None:
         response = ResponseText("Resource not found on this server.", status_code=404)
         await response(scope, receive, send)  # type: ignore

src/reactpy/executors/asgi/pyscript.py

@@ -9,12 +9,12 @@
 from pathlib import Path
 from typing import Any
 
-from asgiref.typing import WebSocketScope
 from typing_extensions import Unpack
 
 from reactpy import html
 from reactpy.executors.asgi.middleware import ReactPyMiddleware
 from reactpy.executors.asgi.standalone import ReactPy, ReactPyApp
+from reactpy.executors.asgi.types import AsgiWebsocketScope
 from reactpy.executors.utils import vdom_head_to_html
 from reactpy.pyscript.utils import pyscript_component_html, pyscript_setup_html
 from reactpy.types import ReactPyConfig, VdomDict
@@ -79,7 +79,9 @@ def __init__(
         self.html_head = html_head or html.head()
         self.html_lang = html_lang
 
-    def match_dispatch_path(self, scope: WebSocketScope) -> bool:  # pragma: no cover
+    def match_dispatch_path(
+        self, scope: AsgiWebsocketScope
+    ) -> bool:  # pragma: no cover
         """We do not use a WebSocket dispatcher for Client-Side Rendering (CSR)."""
         return False
 

src/reactpy/executors/asgi/standalone.py

@@ -9,16 +9,19 @@
 from typing import Callable, Literal, cast, overload
 
 from asgi_tools import ResponseHTML
-from asgiref import typing as asgi_types
 from typing_extensions import Unpack
 
 from reactpy import html
 from reactpy.executors.asgi.middleware import ReactPyMiddleware
 from reactpy.executors.asgi.types import (
     AsgiApp,
-    AsgiHttpApp,
-    AsgiLifespanApp,
-    AsgiWebsocketApp,
+    AsgiReceive,
+    AsgiScope,
+    AsgiSend,
+    AsgiV3HttpApp,
+    AsgiV3LifespanApp,
+    AsgiV3WebsocketApp,
+    AsgiWebsocketScope,
 )
 from reactpy.executors.utils import server_side_component_html, vdom_head_to_html
 from reactpy.pyscript.utils import pyscript_setup_html
@@ -77,11 +80,11 @@ def __init__(
             pyscript_head_vdom["tagName"] = ""
             self.html_head["children"].append(pyscript_head_vdom)  # type: ignore
 
-    def match_dispatch_path(self, scope: asgi_types.WebSocketScope) -> bool:
+    def match_dispatch_path(self, scope: AsgiWebsocketScope) -> bool:
         """Method override to remove `dotted_path` from the dispatcher URL."""
         return str(scope["path"]) == self.dispatcher_path
 
-    def match_extra_paths(self, scope: asgi_types.Scope) -> AsgiApp | None:
+    def match_extra_paths(self, scope: AsgiScope) -> AsgiApp | None:
         """Method override to match user-provided HTTP/Websocket routes."""
         if scope["type"] == "lifespan":
             return self.extra_lifespan_app
@@ -106,22 +109,22 @@ def route(
         self,
         path: str,
         type: Literal["http"] = "http",
-    ) -> Callable[[AsgiHttpApp | str], AsgiApp]: ...
+    ) -> Callable[[AsgiV3HttpApp | str], AsgiApp]: ...
 
     @overload
     def route(
         self,
         path: str,
         type: Literal["websocket"],
-    ) -> Callable[[AsgiWebsocketApp | str], AsgiApp]: ...
+    ) -> Callable[[AsgiV3WebsocketApp | str], AsgiApp]: ...
 
     def route(
         self,
         path: str,
         type: Literal["http", "websocket"] = "http",
     ) -> (
-        Callable[[AsgiHttpApp | str], AsgiApp]
-        | Callable[[AsgiWebsocketApp | str], AsgiApp]
+        Callable[[AsgiV3HttpApp | str], AsgiApp]
+        | Callable[[AsgiV3WebsocketApp | str], AsgiApp]
     ):
         """Interface that allows user to define their own HTTP/Websocket routes
         within the current ReactPy application.
@@ -142,15 +145,15 @@ def decorator(
 
             asgi_app: AsgiApp = import_dotted_path(app) if isinstance(app, str) else app
             if type == "http":
-                self.extra_http_routes[re_path] = cast(AsgiHttpApp, asgi_app)
+                self.extra_http_routes[re_path] = cast(AsgiV3HttpApp, asgi_app)
             elif type == "websocket":
-                self.extra_ws_routes[re_path] = cast(AsgiWebsocketApp, asgi_app)
+                self.extra_ws_routes[re_path] = cast(AsgiV3WebsocketApp, asgi_app)
 
             return asgi_app
 
         return decorator
 
-    def lifespan(self, app: AsgiLifespanApp | str) -> None:
+    def lifespan(self, app: AsgiV3LifespanApp | str) -> None:
         """Interface that allows user to define their own lifespan app
         within the current ReactPy application.
 
@@ -176,10 +179,7 @@ class ReactPyApp:
     _last_modified = ""
 
     async def __call__(
-        self,
-        scope: asgi_types.Scope,
-        receive: asgi_types.ASGIReceiveCallable,
-        send: asgi_types.ASGISendCallable,
+        self, scope: AsgiScope, receive: AsgiReceive, send: AsgiSend
     ) -> None:
         if scope["type"] != "http":  # pragma: no cover
             if scope["type"] != "lifespan":