diff --git a/falcon/_typing.py b/falcon/_typing.py index e84f7c111..a6a360e75 100644 --- a/falcon/_typing.py +++ b/falcon/_typing.py @@ -28,20 +28,21 @@ Any, Callable, Literal, - Optional, Protocol, TYPE_CHECKING, + TypeAlias, + TypedDict, TypeVar, - Union, ) # NOTE(vytas): Mypy still struggles to handle a conditional import in the EAFP # fashion, so we branch on Py version instead (which it does understand). if sys.version_info >= (3, 11): + from typing import NotRequired from wsgiref.types import StartResponse as StartResponse - from wsgiref.types import WSGIEnvironment as WSGIEnvironment else: - WSGIEnvironment = dict[str, Any] + from typing_extensions import NotRequired + StartResponse = Callable[[str, list[tuple[str, str]]], Callable[[bytes], None]] if TYPE_CHECKING: @@ -61,7 +62,96 @@ class _Unset(Enum): _T = TypeVar('_T') _UNSET = _Unset.UNSET -UnsetOr = Union[Literal[_Unset.UNSET], _T] +UnsetOr: TypeAlias = _T | Literal[_Unset.UNSET] + + +# ASGI scope TypedDicts +class _ASGIVersions(TypedDict): + spec_version: str + version: Literal['2.0'] | Literal['3.0'] + + +class HTTPScope(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 + state: NotRequired[dict[str, Any]] + extensions: dict[str, dict[object, object]] | None + + +class WebSocketScope(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] + state: NotRequired[dict[str, Any]] + extensions: dict[str, dict[object, object]] | None + + +class LifespanScope(TypedDict): + type: Literal['lifespan'] + asgi: _ASGIVersions + state: NotRequired[dict[str, Any]] + + +# WSGI environ TypedDict +# NOTE: Keys containing dots (e.g. 'wsgi.input') cannot be expressed in the +# class-based TypedDict syntax; the functional form is used instead so that +# all PEP 3333 keys can be represented in a single type. +_WSGIEnvironmentRequired = TypedDict( + '_WSGIEnvironmentRequired', + { + 'REQUEST_METHOD': str, + 'SCRIPT_NAME': str, + 'PATH_INFO': str, + 'SERVER_NAME': str, + 'SERVER_PORT': str, + 'SERVER_PROTOCOL': str, + 'wsgi.version': tuple[int, int], + 'wsgi.url_scheme': str, + 'wsgi.input': Any, + 'wsgi.errors': Any, + 'wsgi.multithread': bool, + 'wsgi.multiprocess': bool, + 'wsgi.run_once': bool, + }, +) + + +class _WSGIEnvironmentOptional(TypedDict, total=False): + QUERY_STRING: str + CONTENT_TYPE: str + CONTENT_LENGTH: str + REMOTE_ADDR: str + HTTP_HOST: str + HTTP_ACCEPT: str + HTTP_FORWARDED: str + HTTP_X_FORWARDED_FOR: str + HTTP_X_FORWARDED_HOST: str + HTTP_X_FORWARDED_PROTO: str + HTTP_X_REAL_IP: str + + +class WSGIEnvironment(_WSGIEnvironmentRequired, _WSGIEnvironmentOptional): + pass + # NOTE(vytas,jap): TypeVar's "default" argument is only available on 3.13+. if sys.version_info >= (3, 13): @@ -82,7 +172,7 @@ class _Unset(Enum): _ARespT = TypeVar('_ARespT', bound='AsgiResponse', contravariant=True) Link = dict[str, str] -CookieArg = Mapping[str, Union[str, Cookie]] +CookieArg = Mapping[str, str | Cookie] # Error handlers @@ -112,7 +202,7 @@ async def __call__( ErrorSerializer = Callable[[_ReqT, _RespT, 'HTTPError'], None] # Sinks -SinkPrefix = Union[str, Pattern[str]] +SinkPrefix = str | Pattern[str] class SinkCallable(Protocol[_ReqT, _RespT]): @@ -127,11 +217,11 @@ async def __call__( HeaderMapping = Mapping[str, str] HeaderIter = Iterable[tuple[str, str]] -HeaderArg = Union[HeaderMapping, HeaderIter] -ResponseStatus = Union[http.HTTPStatus, str, int] -StoreArg = Optional[dict[str, Any]] +HeaderArg = HeaderMapping | HeaderIter +ResponseStatus = http.HTTPStatus | str | int +StoreArg = dict[str, Any] | None Resource = object -RangeSetHeader = Union[tuple[int, int, int], tuple[int, int, int, str]] +RangeSetHeader = tuple[int, int, int] | tuple[int, int, int, str] # WSGI @@ -151,11 +241,9 @@ def __call__(self, req: Request, resp: Response, **kwargs: Any) -> None: ... ProcessRequestMethod = Callable[['Request', 'Response'], None] ProcessResourceMethod = Callable[ - ['Request', 'Response', Optional[Resource], dict[str, Any]], None -] -ProcessResponseMethod = Callable[ - ['Request', 'Response', Optional[Resource], bool], None + ['Request', 'Response', Resource | None, dict[str, Any]], None ] +ProcessResponseMethod = Callable[['Request', 'Response', Resource | None, bool], None] # ASGI @@ -185,27 +273,27 @@ async def __call__( AsgiSend = Callable[['AsgiSendMsg'], Awaitable[None]] AsgiProcessRequestMethod = Callable[['AsgiRequest', 'AsgiResponse'], Awaitable[None]] AsgiProcessResourceMethod = Callable[ - ['AsgiRequest', 'AsgiResponse', Optional[Resource], dict[str, Any]], Awaitable[None] + ['AsgiRequest', 'AsgiResponse', Resource | None, dict[str, Any]], Awaitable[None] ] AsgiProcessResponseMethod = Callable[ - ['AsgiRequest', 'AsgiResponse', Optional[Resource], bool], Awaitable[None] + ['AsgiRequest', 'AsgiResponse', Resource | None, bool], Awaitable[None] ] AsgiProcessRequestWsMethod = Callable[['AsgiRequest', 'WebSocket'], Awaitable[None]] AsgiProcessResourceWsMethod = Callable[ - ['AsgiRequest', 'WebSocket', Optional[Resource], dict[str, Any]], Awaitable[None] -] -ResponseCallbacks = Union[ - tuple[Callable[[], None], Literal[False]], - tuple[Callable[[], Awaitable[None]], Literal[True]], + ['AsgiRequest', 'WebSocket', Resource | None, dict[str, Any]], Awaitable[None] ] +ResponseCallbacks: TypeAlias = ( + tuple[Callable[[], None], Literal[False]] + | tuple[Callable[[], Awaitable[None]], Literal[True]] +) # Routing -MethodDict = Union[ - dict[str, ResponderCallable], - dict[str, Union[AsgiResponderCallable, AsgiResponderWsCallable]], -] +MethodDict = ( + dict[str, ResponderCallable] + | dict[str, AsgiResponderCallable | AsgiResponderWsCallable] +) class FindMethod(Protocol): @@ -221,7 +309,7 @@ def __call__(self, media: Any, content_type: str | None = ...) -> bytes: ... DeserializeSync = Callable[[bytes], Any] -Responder = Union[ResponderMethod, AsgiResponderMethod] +Responder = ResponderMethod | AsgiResponderMethod # WSGI middleware interface @@ -356,32 +444,35 @@ async def process_response_async( # NOTE(jkmnt): This typing is far from perfect due to the Python typing limitations, # but better than nothing. Middleware conforming to any protocol of the union # will pass the type check. Other protocols violations are not checked. -SyncMiddleware = Union[ - WsgiMiddlewareWithProcessRequest[_ReqT, _RespT], - WsgiMiddlewareWithProcessResource[_ReqT, _RespT], - WsgiMiddlewareWithProcessResponse[_ReqT, _RespT], -] +SyncMiddleware = ( + WsgiMiddlewareWithProcessRequest[_ReqT, _RespT] + | WsgiMiddlewareWithProcessResource[_ReqT, _RespT] + | WsgiMiddlewareWithProcessResponse[_ReqT, _RespT] +) """Synchronous (WSGI) application middleware. This type alias reflects the middleware interface for components that can be used with a WSGI app. """ -AsyncMiddleware = Union[ - AsgiMiddlewareWithProcessRequest[_AReqT, _ARespT], - AsgiMiddlewareWithProcessResource[_AReqT, _ARespT], - AsgiMiddlewareWithProcessResponse[_AReqT, _ARespT], +AsyncMiddleware = ( + AsgiMiddlewareWithProcessRequest[_AReqT, _ARespT] + | AsgiMiddlewareWithProcessResource[_AReqT, _ARespT] + | AsgiMiddlewareWithProcessResponse[_AReqT, _ARespT] + | # Lifespan middleware - AsgiMiddlewareWithProcessStartup, - AsgiMiddlewareWithProcessShutdown, + AsgiMiddlewareWithProcessStartup + | AsgiMiddlewareWithProcessShutdown + | # WebSocket middleware - AsgiMiddlewareWithProcessRequestWs[_AReqT], - AsgiMiddlewareWithProcessResourceWs[_AReqT], + AsgiMiddlewareWithProcessRequestWs[_AReqT] + | AsgiMiddlewareWithProcessResourceWs[_AReqT] + | # Universal middleware with process_*_async methods - UniversalMiddlewareWithProcessRequest[_AReqT, _ARespT], - UniversalMiddlewareWithProcessResource[_AReqT, _ARespT], - UniversalMiddlewareWithProcessResponse[_AReqT, _ARespT], -] + UniversalMiddlewareWithProcessRequest[_AReqT, _ARespT] + | UniversalMiddlewareWithProcessResource[_AReqT, _ARespT] + | UniversalMiddlewareWithProcessResponse[_AReqT, _ARespT] +) """Asynchronous (ASGI) application middleware. This type alias reflects the middleware interface for components that can be diff --git a/falcon/app.py b/falcon/app.py index 3b97e87bd..eb1972000 100644 --- a/falcon/app.py +++ b/falcon/app.py @@ -499,7 +499,7 @@ def __call__( # noqa: C901 length: int | None = 0 try: - body, length = self._get_body(resp, env.get('wsgi.file_wrapper')) + body, length = self._get_body(resp, env.get('wsgi.file_wrapper')) # type: ignore[arg-type] except Exception as ex: if not self._handle_exception(req, resp, ex, params): raise diff --git a/falcon/asgi/app.py b/falcon/asgi/app.py index 375c252ea..ea2f2c411 100644 --- a/falcon/asgi/app.py +++ b/falcon/asgi/app.py @@ -45,8 +45,10 @@ from falcon._typing import AsgiSend from falcon._typing import AsgiSinkCallable 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 @@ -466,16 +468,19 @@ 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'] + raw = scope['asgi'] except KeyError: - # NOTE(kgriffs): According to the ASGI spec, "2.0" is - # the default version. - asgi_info = scope['asgi'] = {'version': '2.0'} + # Default per ASGI spec + raw = {'version': '2.0'} + scope['asgi'] = raw - try: - spec_version: str | None = asgi_info['spec_version'] - except KeyError: - spec_version = None + # Normalize into proper _ASGIVersions shape + asgi_info = { + 'spec_version': str(raw.get('spec_version', '2.0')), + 'version': str(raw.get('version', '2.0')), + } + + spec_version: str | None = asgi_info['spec_version'] try: http_version: str = scope['http_version'] @@ -488,12 +493,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 +519,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 +1126,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 +1135,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: dict[str, str] = scope['asgi'] # type: ignore[assignment] version = asgi_info.get('version', '2.0 (implicit)') if not version.startswith('3.'): await send( @@ -1188,7 +1196,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..f8680da0b 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,7 +101,7 @@ 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`_). @@ -111,7 +113,7 @@ class Request(request.Request): def __init__( self, - scope: dict[str, Any], + scope: HTTPScope | WebSocketScope, receive: AsgiReceive, first_event: AsgiEvent | None = None, options: request.RequestOptions | None = None, @@ -160,7 +162,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. @@ -346,7 +348,7 @@ def root_path(self) -> str: # 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] + return self.scope['root_path'] except KeyError: pass @@ -381,7 +383,7 @@ def scheme(self) -> str: # 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] + return self.scope['scheme'] except KeyError: pass @@ -493,24 +495,12 @@ def access_route(self) -> list[str]: if self._cached_access_route is None: # PERF(kgriffs): 'client' is optional according to the ASGI spec # but it will probably be present, hence the try...except. - try: - # NOTE(kgriffs): The ASGI spec states that this can be - # any iterable. So we need to read and cache it in - # 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'] - # 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 - # it must be a valid iterable when present. In any case, we - # simply catch TypeError here too to account for this scenario. - except (KeyError, TypeError): - # NOTE(kgriffs): Default to localhost so that app logic does - # note have to special-case the handling of a missing - # client field in the connection scope. This should be - # a reasonable default, but we can change it later if - # that turns out not to be the case. + + client_info = self.scope.get('client') + + if client_info is not None: + client, __ = client_info + else: client = '127.0.0.1' headers = self._asgi_headers @@ -519,7 +509,7 @@ def access_route(self) -> list[str]: self._cached_access_route = [] for hop in self.forwarded or (): if hop.src is not None: - host, __ = parse_host(hop.src) + host = parse_host(hop.src)[0] self._cached_access_route.append(host) elif b'x-forwarded-for' in headers: addresses = headers[b'x-forwarded-for'].decode('latin1').split(',') @@ -920,7 +910,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 # pragma: no cover + host, port = server + self._asgi_server_cached = ( + str(host), + int(port) + if port is not None + else (443 if self._secure_scheme else 80), + ) except (KeyError, TypeError): # NOTE(kgriffs): Not found, or was None default_port = 443 if self._secure_scheme else 80 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/request.py b/falcon/request.py index ec4d77148..a2fe7a067 100644 --- a/falcon/request.py +++ b/falcon/request.py @@ -37,6 +37,7 @@ from falcon._typing import _UNSET from falcon._typing import StoreArg from falcon._typing import UnsetOr +from falcon._typing import WSGIEnvironment from falcon.constants import DEFAULT_MEDIA_TYPE from falcon.constants import FALSE_STRINGS from falcon.constants import MEDIA_JSON @@ -140,7 +141,7 @@ class Request: """ # Attribute declaration - env: dict[str, Any] + env: WSGIEnvironment """Reference to the WSGI environ ``dict`` passed in from the server. (See also PEP-3333.) """ @@ -243,7 +244,7 @@ class Request: """Always ``False`` in a sync ``Request``.""" def __init__( - self, env: dict[str, Any], options: RequestOptions | None = None + self, env: WSGIEnvironment, options: RequestOptions | None = None ) -> None: self.is_websocket: bool = False @@ -260,7 +261,6 @@ def __init__( # NOTE(kgriffs): PEP 3333 specifies that PATH_INFO may be the # empty string, so normalize it in that case. path: str = env['PATH_INFO'] or '/' - # PEP 3333 specifies that the PATH_INFO variable is always # "bytes tunneled as latin-1" and must be encoded back. # @@ -503,7 +503,7 @@ def if_match(self) -> list[ETag | Literal['*']] | None: if self._cached_if_match is _UNSET: header_value = self.env.get('HTTP_IF_MATCH') if header_value: - self._cached_if_match = helpers._parse_etags(header_value) + self._cached_if_match = helpers._parse_etags(str(header_value)) else: self._cached_if_match = None @@ -524,7 +524,7 @@ def if_none_match(self) -> list[ETag | Literal['*']] | None: if self._cached_if_none_match is _UNSET: header_value = self.env.get('HTTP_IF_NONE_MATCH') if header_value: - self._cached_if_none_match = helpers._parse_etags(header_value) + self._cached_if_none_match = helpers._parse_etags(str(header_value)) else: self._cached_if_none_match = None @@ -647,7 +647,7 @@ def root_path(self) -> str: # include it even in that case. try: # TODO(0xMattB): Implement advanced typing to type as 'str' (see PR #2599) - return self.env['SCRIPT_NAME'] # type: ignore[no-any-return] + return self.env['SCRIPT_NAME'] except KeyError: return '' @@ -673,7 +673,7 @@ def scheme(self) -> str: to handle such cases. """ # TODO(0xMattB): Implement advanced typing to type as 'str' (see PR #2599) - return self.env['wsgi.url_scheme'] # type: ignore[no-any-return] + return self.env['wsgi.url_scheme'] @property def forwarded_scheme(self) -> str: @@ -897,10 +897,10 @@ def headers(self) -> Mapping[str, str]: # NOTE(kgriffs): Don't take the time to fix the case # since headers are supposed to be case-insensitive # anyway. - headers[name[5:].replace('_', '-')] = value + headers[name[5:].replace('_', '-')] = str(value) elif name in WSGI_CONTENT_HEADERS: - headers[name.replace('_', '-')] = value + headers[name.replace('_', '-')] = str(value) return self._cached_headers @@ -1347,7 +1347,7 @@ def get_header( # This will be faster, assuming that most headers are looked # up only once, and not all headers will be requested. # TODO(0xMattB): Implement advanced typing to type as 'str' (see PR #2599) - return self.env['HTTP_' + wsgi_name] # type: ignore[no-any-return] + return str(self.env['HTTP_' + wsgi_name]) # type: ignore[literal-required] except KeyError: # NOTE(kgriffs): There are a couple headers that do not @@ -1358,7 +1358,7 @@ def get_header( try: # TODO(0xMattB): Implement advanced typing to type as 'str' # (see PR #2599). - return self.env[wsgi_name] # type: ignore[no-any-return] + return str(self.env[wsgi_name]) # type: ignore[literal-required] except KeyError: pass diff --git a/falcon/request_helpers.py b/falcon/request_helpers.py index 5e13c8801..be4809b7d 100644 --- a/falcon/request_helpers.py +++ b/falcon/request_helpers.py @@ -20,6 +20,7 @@ import re from typing import Any, Literal, TYPE_CHECKING +from falcon._typing import WSGIEnvironment from falcon.util import ETag if TYPE_CHECKING: @@ -116,10 +117,9 @@ def _header_property(wsgi_name: str) -> Any: """ def fget(self: Request) -> str | None: - try: - return self.env[wsgi_name] or None - except KeyError: - return None + + env: WSGIEnvironment = self.env + return env.get(wsgi_name) or None # type: ignore[return-value] return property(fget) diff --git a/falcon/testing/helpers.py b/falcon/testing/helpers.py index f1ef3ccba..d8c56ee1f 100644 --- a/falcon/testing/helpers.py +++ b/falcon/testing/helpers.py @@ -1315,7 +1315,7 @@ def create_req( """ env = create_environ(**kwargs) - return falcon.request.Request(env, options=options) + return falcon.request.Request(env, options=options) # type: ignore[arg-type] def create_asgi_req( @@ -1354,7 +1354,7 @@ def create_asgi_req( req_event_emitter = ASGIRequestEventEmitter(body, disconnect_at=disconnect_at) req_type = req_type or falcon.asgi.Request - return req_type(scope, req_event_emitter, options=options) + return req_type(scope, req_event_emitter, options=options) # type: ignore[arg-type] # NOTE(TudorGR): Deprecated in Falcon 4.3.