diff --git a/docs/_newsfragments/2628.newandimproved.rst b/docs/_newsfragments/2628.newandimproved.rst new file mode 100644 index 000000000..405d45ba4 --- /dev/null +++ b/docs/_newsfragments/2628.newandimproved.rst @@ -0,0 +1 @@ +Improve ASGI typing by introducing TypedDict definitions for HTTP event messages. \ No newline at end of file diff --git a/falcon/asgi/app.py b/falcon/asgi/app.py index 375c252ea..40ba49bec 100644 --- a/falcon/asgi/app.py +++ b/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) diff --git a/falcon/asgi_spec.py b/falcon/asgi_spec.py index a6cc7658a..b4d442f98 100644 --- a/falcon/asgi_spec.py +++ b/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 \ No newline at end of file