Enhance ASGI scope typing and add HTTP event types

docs/_newsfragments/2628.newandimproved.rst

@@ -0,0 +1 @@
+Improve ASGI typing by introducing TypedDict definitions for HTTP event messages.

falcon/asgi/app.py

@@ -28,6 +28,7 @@
     ClassVar,
     overload,
     TYPE_CHECKING,
+    cast
 )
 import warnings
 
@@ -55,6 +56,7 @@
 from falcon.asgi_spec import AsgiSendMsg
 from falcon.asgi_spec import EventType
 from falcon.asgi_spec import WSCloseCode
+from falcon.asgi_spec import ASGIScope
 from falcon.constants import MEDIA_JSON
 from falcon.errors import CompatibilityError
 from falcon.errors import HTTPBadRequest
@@ -68,7 +70,6 @@
 from falcon.util.sync import _should_wrap_non_coroutines
 from falcon.util.sync import _wrap_non_coroutine_unsafe
 from falcon.util.sync import wrap_sync_to_async
-
 from ._asgi_helpers import _validate_asgi_scope
 from ._asgi_helpers import _wrap_asgi_coroutine_func
 from .multipart import MultipartForm
@@ -456,7 +457,7 @@ def __init__(
     @_wrap_asgi_coroutine_func
     async def __call__(  # type: ignore[override] # noqa: C901
         self,
-        scope: dict[str, Any],
+        scope: ASGIScope | dict[str, Any],
         receive: AsgiReceive,
         send: AsgiSend,
     ) -> None:
@@ -488,12 +489,12 @@ async def __call__(  # type: ignore[override] # noqa: C901
             # PERF(vytas): Evaluate the potentially recurring WebSocket path
             #   first (in contrast to one-shot lifespan events).
             if scope_type == 'websocket':
-                await self._handle_websocket(spec_version, scope, receive, send)
+                await self._handle_websocket(spec_version, cast(dict[str, Any], scope), receive, send)
                 return
 
             # NOTE(vytas): Else 'lifespan' -- other scope_type values have been
             #   eliminated by _validate_asgi_scope at this point.
-            await self._call_lifespan_handlers(spec_version, scope, receive, send)
+            await self._call_lifespan_handlers(spec_version, cast(dict[str, Any], scope), receive, send)
             return
 
         # NOTE(kgriffs): Per the ASGI spec, we should not proceed with request
@@ -514,7 +515,7 @@ async def __call__(  # type: ignore[override] # noqa: C901
         assert first_event_type == 'http.request'
 
         req = self._request_type(
-            scope, receive, first_event=first_event, options=self.req_options
+            cast(dict[str, Any], scope), receive, first_event=first_event, options=self.req_options
         )
         resp = self._response_type(options=self.resp_options)
 

falcon/asgi_spec.py

@@ -15,9 +15,10 @@
 """Constants, etc. defined by the ASGI specification."""
 
 from __future__ import annotations
-
 from collections.abc import Mapping
-from typing import Any
+
+from typing import TypedDict, Literal,  Any, Required
+
 
 
 class EventType:
@@ -42,6 +43,45 @@ class EventType:
     WS_DISCONNECT = 'websocket.disconnect'
     WS_CLOSE = 'websocket.close'
 
+class HTTPRequestEvent(TypedDict, total=False):
+    type: Literal["http.request"]
+    body: bytes
+    more_body: bool
+
+class HTTPDisconnectEvent(TypedDict):
+    type: Literal["http.disconnect"]
+
+class HTTPResponseStartEvent(TypedDict):
+    type: Literal["http.response.start"]
+    status: int
+    headers: list[tuple[bytes, bytes]]
+
+
+class HTTPResponseBodyEvent(TypedDict, total=False):
+    type: Literal["http.response.body"]
+    body: bytes
+    more_body: bool
+
+class HTTPScope(TypedDict, total=False):
+    type: Required[Literal["http"]]
+    http_version: str
+    method: str
+    path: str
+    headers: list[tuple[bytes, bytes]]
+    asgi: dict[str, Any]
+
+class WebSocketScope(TypedDict, total=False):
+    type: Required[Literal["websocket"]]
+    path: str
+    headers: list[tuple[bytes, bytes]]
+    http_version: str
+    asgi: dict[str, Any]
+
+class LifespanScope(TypedDict, total=False):
+    type: Required[Literal["lifespan"]]
+    asgi: dict[str, Any]
+    http_version: str
+
 
 class ScopeType:
     """Standard ASGI event type strings."""
@@ -64,7 +104,6 @@ class WSCloseCode:
     HANDLER_NOT_FOUND = 3405
 
 
-# TODO: use a typed dict for event dicts
 AsgiEvent = Mapping[str, Any]
-# TODO: use a typed dict for send msg dicts
 AsgiSendMsg = dict[str, Any]
+ASGIScope = HTTPScope | WebSocketScope | LifespanScope