From 173af78e778177802b546bec17414106091d3d95 Mon Sep 17 00:00:00 2001 From: smtalha682 Date: Thu, 11 Jun 2026 00:53:33 +0530 Subject: [PATCH] feat(typing): added type hints for ASGI Scope dict --- falcon/_typing.py | 77 +++++++++++++++++++++++++++++++++++++++ falcon/asgi/app.py | 25 ++++++++----- falcon/asgi/request.py | 35 +++++++++++------- falcon/asgi/ws.py | 3 +- falcon/testing/client.py | 12 +++--- falcon/testing/helpers.py | 50 ++++++++++++++++++------- 6 files changed, 159 insertions(+), 43 deletions(-) diff --git a/falcon/_typing.py b/falcon/_typing.py index 56d66b6f1..03604ca06 100644 --- a/falcon/_typing.py +++ b/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 diff --git a/falcon/asgi/app.py b/falcon/asgi/app.py index 375c252ea..acdce5b81 100644 --- a/falcon/asgi/app.py +++ b/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: diff --git a/falcon/asgi/request.py b/falcon/asgi/request.py index 807153ce5..cd4f46220 100644 --- a/falcon/asgi/request.py +++ b/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: diff --git a/falcon/asgi/ws.py b/falcon/asgi/ws.py index 52192974f..308a76611 100644 --- a/falcon/asgi/ws.py +++ b/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[ diff --git a/falcon/testing/client.py b/falcon/testing/client.py index 3259be5f8..e15ad6cb3 100644 --- a/falcon/testing/client.py +++ b/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', diff --git a/falcon/testing/helpers.py b/falcon/testing/helpers.py index f1ef3ccba..7ea3415b9 100644 --- a/falcon/testing/helpers.py +++ b/falcon/testing/helpers.py @@ -51,11 +51,12 @@ from falcon import errors as falcon_errors from falcon._typing import CookieArg from falcon._typing import HeaderArg +from falcon._typing import HTTPScope from falcon._typing import ResponseStatus +from falcon._typing import WebSocketScope import falcon.asgi from falcon.asgi_spec import AsgiEvent from falcon.asgi_spec import EventType -from falcon.asgi_spec import ScopeType from falcon.asgi_spec import WSCloseCode from falcon.constants import SINGLETON_HEADERS import falcon.request @@ -911,7 +912,7 @@ def create_scope( content_length: int | None = None, include_server: bool = True, cookies: CookieArg | None = None, -) -> dict[str, Any]: +) -> HTTPScope: """Create a mock ASGI scope ``dict`` for simulating HTTP requests. Keyword Args: @@ -979,8 +980,8 @@ def create_scope( if query_string_bytes and query_string_bytes.startswith(b'?'): raise ValueError("query_string should not start with '?'") - scope: dict[str, Any] = { - 'type': ScopeType.HTTP, + scope: HTTPScope = { + 'type': 'http', 'asgi': { 'version': '3.0', 'spec_version': '2.1', @@ -990,6 +991,11 @@ def create_scope( 'path': path, 'raw_path': raw_path.encode(), 'query_string': query_string_bytes, + 'root_path': '', + 'scheme': 'http', + 'headers': [], + 'client': None, + 'server': None, } # NOTE(kgriffs): Explicitly test against None so that the caller @@ -1025,10 +1031,10 @@ def create_scope( # NOTE(kgriffs): Expose as an iterable to ensure the framework/app # isn't hard-coded to only work with a list or tuple. - scope['client'] = iter([remote_addr, remote_port]) + scope['client'] = iter([remote_addr, remote_port]) # type: ignore[typeddict-item] if include_server: - scope['server'] = iter([host, port]) + scope['server'] = iter([host, port]) # type: ignore[typeddict-item] # NOTE(myusko): Clients discard Set-Cookie header # in the response to the OPTIONS method. @@ -1055,7 +1061,7 @@ def create_scope_ws( include_server: bool = True, subprotocols: str | None = None, spec_version: str = '2.1', -) -> dict[str, Any]: +) -> WebSocketScope: """Create a mock ASGI scope ``dict`` for simulating WebSocket requests. Keyword Args: @@ -1100,7 +1106,11 @@ def create_scope_ws( advertise to the server (default ``[]``). """ - scope = create_scope( + # NOTE(symtalha14): Build from the HTTP scope to reuse all its construction + # logic, then copy shared fields into a proper WebSocketScope. We cannot + # mutate the HTTPScope in-place because TypedDicts do not support deleting + # required keys (e.g. 'method'). + http_scope = create_scope( path=path, query_string=query_string, headers=headers, @@ -1113,9 +1123,23 @@ def create_scope_ws( include_server=include_server, ) - scope['type'] = ScopeType.WS - scope['asgi']['spec_version'] = spec_version - del scope['method'] + scope: WebSocketScope = { + 'type': 'websocket', + 'asgi': { + 'version': http_scope['asgi']['version'], + 'spec_version': spec_version, + }, + 'http_version': http_scope['http_version'], + 'scheme': http_scope.get('scheme', 'ws'), + 'path': http_scope['path'], + 'raw_path': http_scope['raw_path'], + 'query_string': http_scope['query_string'], + 'root_path': http_scope.get('root_path', ''), + 'headers': http_scope.get('headers', []), + 'client': http_scope.get('client'), + 'server': http_scope.get('server'), + 'subprotocols': [], + } # NOTE(kgriffiths): Explicit check against None affords simulating a request # with a scope that does not contain the optional 'subprotocols' key. @@ -1479,7 +1503,7 @@ def _add_headers_to_environ(env: dict[str, Any], headers: HeaderArg | None) -> N def _add_headers_to_scope( - scope: dict[str, Any], + scope: HTTPScope | WebSocketScope, headers: HeaderArg | None, content_length: int | None, host: str, @@ -1533,7 +1557,7 @@ def _add_headers_to_scope( # NOTE(kgriffs): Make it an iterator to ensure the app is not expecting # a specific type (ASGI only specified that it is an iterable). - scope['headers'] = iter(prepared_headers) + scope['headers'] = iter(prepared_headers) # type: ignore[arg-type] def _fixup_http_version(http_version: str) -> str: