feat(typing): added type hints for ASGI Scope dict

falcon/_typing.py

@@ -31,6 +31,7 @@
     Optional,
     Protocol,
     TYPE_CHECKING,
+    TypedDict,
     TypeVar,
     Union,
 )
@@ -44,6 +45,82 @@
     WSGIEnvironment = dict[str, Any]
     StartResponse = Callable[[str, list[tuple[str, str]]], Callable[[bytes], None]]
 
+
+# ---------------------------------------------------------------------------
+# ASGI scope TypedDicts
+# Modelled after the ASGI spec:
+#   https://asgi.readthedocs.io/en/latest/specs/www.html
+# ---------------------------------------------------------------------------
+
+
+class _ASGIVersionsRequired(TypedDict):
+    version: str
+
+
+class ASGIVersions(_ASGIVersionsRequired, total=False):
+    """ASGI version info nested inside every scope dict."""
+
+    spec_version: str
+
+
+class _HTTPScopeRequired(TypedDict):
+    type: Literal['http']
+    asgi: ASGIVersions
+    http_version: str
+    method: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[tuple[bytes, bytes]]
+    client: tuple[str, int] | None
+    server: tuple[str, int | None] | None
+
+
+class HTTPScope(_HTTPScopeRequired, total=False):
+    """ASGI connection scope for HTTP requests."""
+
+    state: dict[str, Any]
+    extensions: dict[str, dict[object, object]] | None
+
+
+class _WebSocketScopeRequired(TypedDict):
+    type: Literal['websocket']
+    asgi: ASGIVersions
+    http_version: str
+    scheme: str
+    path: str
+    raw_path: bytes
+    query_string: bytes
+    root_path: str
+    headers: Iterable[tuple[bytes, bytes]]
+    client: tuple[str, int] | None
+    server: tuple[str, int | None] | None
+    subprotocols: Iterable[str]
+
+
+class WebSocketScope(_WebSocketScopeRequired, total=False):
+    """ASGI connection scope for WebSocket connections."""
+
+    state: dict[str, Any]
+    extensions: dict[str, dict[object, object]] | None
+
+
+class _LifespanScopeRequired(TypedDict):
+    type: Literal['lifespan']
+    asgi: ASGIVersions
+
+
+class LifespanScope(_LifespanScopeRequired, total=False):
+    """ASGI connection scope for lifespan events."""
+
+    state: dict[str, Any]
+
+
+# Union of all scope types -- use when scope type has not yet been narrowed.
+AsgiScope = Union[HTTPScope, WebSocketScope, LifespanScope]
+
 if TYPE_CHECKING:
     from falcon.asgi import Request as AsgiRequest
     from falcon.asgi import Response as AsgiResponse

falcon/asgi/app.py

@@ -42,11 +42,15 @@
 from falcon._typing import AsgiReceive
 from falcon._typing import AsgiResponderCallable
 from falcon._typing import AsgiResponderWsCallable
+from falcon._typing import AsgiScope
 from falcon._typing import AsgiSend
 from falcon._typing import AsgiSinkCallable
+from falcon._typing import ASGIVersions
 from falcon._typing import AsyncMiddleware
+from falcon._typing import LifespanScope
 from falcon._typing import Resource
 from falcon._typing import SinkPrefix
+from falcon._typing import WebSocketScope
 import falcon.app
 from falcon.app_helpers import AsyncPreparedMiddlewareResult
 from falcon.app_helpers import AsyncPreparedMiddlewareWsResult
@@ -456,7 +460,7 @@ def __init__(
     @_wrap_asgi_coroutine_func
     async def __call__(  # type: ignore[override] # noqa: C901
         self,
-        scope: dict[str, Any],
+        scope: AsgiScope,
         receive: AsgiReceive,
         send: AsgiSend,
     ) -> None:
@@ -466,7 +470,7 @@ async def __call__(  # type: ignore[override] # noqa: C901
         # PERF(kgriffs): This should usually be present, so use a
         #   try..except
         try:
-            asgi_info: dict[str, str] = scope['asgi']
+            asgi_info: ASGIVersions = scope['asgi']
         except KeyError:
             # NOTE(kgriffs): According to the ASGI spec, "2.0" is
             #   the default version.
@@ -478,7 +482,7 @@ async def __call__(  # type: ignore[override] # noqa: C901
             spec_version = None
 
         try:
-            http_version: str = scope['http_version']
+            http_version: str = scope['http_version']  # type: ignore[typeddict-item]
         except KeyError:
             http_version = '1.1'
 
@@ -488,12 +492,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, scope, receive, send)  # type: ignore[arg-type]
                 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, scope, receive, send)  # type: ignore[arg-type]
             return
 
         # NOTE(kgriffs): Per the ASGI spec, we should not proceed with request
@@ -514,7 +518,10 @@ 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
+            scope,  # type: ignore[arg-type]
+            receive,
+            first_event=first_event,
+            options=self.req_options,
         )
         resp = self._response_type(options=self.resp_options)
 
@@ -1118,7 +1125,7 @@ def _schedule_callbacks(self, resp: Response) -> None:
                 loop.run_in_executor(None, cb)
 
     async def _call_lifespan_handlers(
-        self, ver: str, scope: dict[str, Any], receive: AsgiReceive, send: AsgiSend
+        self, ver: str, scope: LifespanScope, receive: AsgiReceive, send: AsgiSend
     ) -> None:
         while True:
             event = await receive()
@@ -1127,7 +1134,7 @@ async def _call_lifespan_handlers(
                 #   startup, as opposed to repeating them every request.
 
                 # NOTE(vytas): If missing, 'asgi' is populated in __call__.
-                asgi_info: dict[str, str] = scope['asgi']
+                asgi_info: ASGIVersions = scope['asgi']
                 version = asgi_info.get('version', '2.0 (implicit)')
                 if not version.startswith('3.'):
                     await send(
@@ -1188,7 +1195,7 @@ async def _call_lifespan_handlers(
                 return
 
     async def _handle_websocket(
-        self, ver: str, scope: dict[str, Any], receive: AsgiReceive, send: AsgiSend
+        self, ver: str, scope: WebSocketScope, receive: AsgiReceive, send: AsgiSend
     ) -> None:
         first_event = await receive()
         if first_event['type'] != EventType.WS_CONNECT:

falcon/asgi/request.py

@@ -31,8 +31,10 @@
 from falcon import request_helpers as helpers
 from falcon._typing import _UNSET
 from falcon._typing import AsgiReceive
+from falcon._typing import HTTPScope
 from falcon._typing import StoreArg
 from falcon._typing import UnsetOr
+from falcon._typing import WebSocketScope
 from falcon.asgi_spec import AsgiEvent
 from falcon.constants import SINGLETON_HEADERS
 from falcon.forwarded import Forwarded
@@ -99,19 +101,23 @@ class Request(request.Request):
     _media_error: Exception | None = None
     _stream: BoundedStream | None = None
 
-    scope: dict[str, Any]
+    scope: HTTPScope | WebSocketScope
     """Reference to the ASGI HTTP connection scope passed in
     from the server (see also: `Connection Scope`_).
 
     .. _Connection Scope:
         https://asgi.readthedocs.io/en/latest/specs/www.html#connection-scope
+
+    .. versionchanged:: 4.3
+        The type was narrowed from ``dict[str, Any]`` to
+        ``HTTPScope | WebSocketScope``.
     """
     is_websocket: bool
     """Set to ``True`` IFF this request was made as part of a WebSocket handshake."""
 
     def __init__(
         self,
-        scope: dict[str, Any],
+        scope: HTTPScope | WebSocketScope,
         receive: AsgiReceive,
         first_event: AsgiEvent | None = None,
         options: request.RequestOptions | None = None,
@@ -160,7 +166,7 @@ def __init__(
 
         self.options = options if options is not None else request.RequestOptions()
 
-        self.method = 'GET' if self.is_websocket else scope['method']
+        self.method = 'GET' if self.is_websocket else scope['method']  # type: ignore[typeddict-item]
 
         self.uri_template = None
         # PERF(vytas): Fall back to class variable(s) when unset.
@@ -345,12 +351,11 @@ def root_path(self) -> str:
         #   empty string, at least uvicorn still includes it explicitly in
         #   that case.
         try:
-            # TODO(0xMattB): Implement advanced typing to type as 'str' (see gh #2628).
-            return self.scope['root_path']  # type: ignore[no-any-return]
-        except KeyError:
+            return self.scope['root_path']
+        except KeyError:  # pragma: nocover
             pass
 
-        return ''
+        return ''  # pragma: nocover
 
     @property
     # NOTE(caselit): Deprecated long ago. Warns since 4.0.
@@ -380,12 +385,11 @@ def scheme(self) -> str:
         # PERF(kgriffs): Use try...except because we normally expect the
         #   key to be present.
         try:
-            # TODO(0xMattB): Implement advanced typing to type as 'str' (see gh #2628).
-            return self.scope['scheme']  # type: ignore[no-any-return]
-        except KeyError:
+            return self.scope['scheme']
+        except KeyError:  # pragma: nocover
             pass
 
-        return 'ws' if self.is_websocket else 'http'
+        return 'ws' if self.is_websocket else 'http'  # pragma: nocover
 
     @property
     def forwarded_scheme(self) -> str:
@@ -499,7 +503,7 @@ def access_route(self) -> list[str]:
                 #   case the iterable is forward-only. But that is
                 #   effectively what we are doing since we only ever
                 #   access this field when setting self._cached_access_route
-                client, __ = self.scope['client']
+                client, __ = self.scope['client']  # type: ignore[misc]
             # NOTE(vytas): Uvicorn may explicitly set scope['client'] to None.
             #   According to the spec, it does default to None when missing,
             #   but it is unclear whether it can be explicitly set to None, or
@@ -920,13 +924,16 @@ def _asgi_server(self) -> tuple[str, int]:
                 #   read it once and cache the result in case the
                 #   iterator is forward-only (not likely, but better
                 #   safe than sorry).
-                self._asgi_server_cached = tuple(self.scope['server'])
+                server = self.scope['server']
+                if server is None:
+                    raise TypeError
+                self._asgi_server_cached = tuple(server)  # type: ignore[assignment]
             except (KeyError, TypeError):
                 # NOTE(kgriffs): Not found, or was None
                 default_port = 443 if self._secure_scheme else 80
                 self._asgi_server_cached = ('localhost', default_port)
 
-        return self._asgi_server_cached
+        return self._asgi_server_cached  # type: ignore[return-value]
 
     @property
     def _secure_scheme(self) -> bool:

falcon/asgi/ws.py

@@ -15,6 +15,7 @@
 from falcon._typing import AsgiReceive
 from falcon._typing import AsgiSend
 from falcon._typing import HeaderArg
+from falcon._typing import WebSocketScope
 from falcon.asgi_spec import AsgiEvent
 from falcon.asgi_spec import AsgiSendMsg
 from falcon.asgi_spec import EventType
@@ -65,7 +66,7 @@ class WebSocket:
     def __init__(
         self,
         ver: str,
-        scope: dict[str, Any],
+        scope: WebSocketScope,
         receive: AsgiReceive,
         send: AsgiSend,
         media_handlers: Mapping[

falcon/testing/client.py

@@ -48,8 +48,8 @@
 from falcon._typing import HeaderArg
 from falcon._typing import HeaderIter
 from falcon._typing import HeaderMapping
+from falcon._typing import LifespanScope
 from falcon.asgi_spec import AsgiEvent
-from falcon.asgi_spec import ScopeType
 from falcon.constants import COMBINED_METHODS
 from falcon.constants import MEDIA_JSON
 from falcon.constants import MEDIA_MSGPACK
@@ -924,7 +924,7 @@ async def _simulate_request_asgi(
             'Please use the method parameter.'
         )
 
-    http_scope.update(extras)
+    http_scope.update(extras)  # type: ignore[typeddict-item]
     # ---------------------------------------------------------------------
 
     if asgi_disconnect_ttl == 0:  # Special case
@@ -971,8 +971,8 @@ async def _simulate_request_asgi(
     # ---------------------------------------------------------------------
     # NOTE(kgriffs): 'lifespan' scope
     # ---------------------------------------------------------------------
-    lifespan_scope = {
-        'type': ScopeType.LIFESPAN,
+    lifespan_scope: LifespanScope = {
+        'type': 'lifespan',
         'asgi': {
             'version': '3.0',
             'spec_version': '2.0',
@@ -1130,8 +1130,8 @@ def __init__(
         self._lifespan_task: asyncio.Task[Any] | None = None
 
     async def __aenter__(self) -> ASGIConductor:
-        lifespan_scope = {
-            'type': ScopeType.LIFESPAN,
+        lifespan_scope: LifespanScope = {
+            'type': 'lifespan',
             'asgi': {
                 'version': '3.0',
                 'spec_version': '2.0',