Skip to content

Improve typing with structured ASGI scopes and WSGI environment TypedDicts #2642

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
MustafaNazir11 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Improve typing with structured ASGI scopes and WSGI environment TypedDicts #2642

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
179 changes: 135 additions & 44 deletions falcon/_typing.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Copy link
Copy Markdown
Member

@vytas7 vytas7 May 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why did you remove WSGIEnvironment from here?

else:
WSGIEnvironment = dict[str, Any]
from typing_extensions import NotRequired

StartResponse = Callable[[str, list[tuple[str, str]]], Callable[[bytes], None]]

if TYPE_CHECKING:
Expand All @@ -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):
Expand All @@ -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
Expand Down Expand Up @@ -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]):
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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):
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion falcon/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
36 changes: 22 additions & 14 deletions falcon/asgi/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Copy link
Copy Markdown
Member

@vytas7 vytas7 May 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please don't remove the notes or make unrelated changes to code.

# 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
Copy link
Copy Markdown
Member

@vytas7 vytas7 May 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please don't make changes that can affect performance in this performance-critical path.

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']
Expand All @@ -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]
Copy link
Copy Markdown
Member

@vytas7 vytas7 May 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The whole point was to get rid of type-ignores, not to add new ones!

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
Expand All @@ -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]
Copy link
Copy Markdown
Member

@vytas7 vytas7 May 4, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here, no new ignores!

receive,
first_event=first_event,
options=self.req_options,
)
resp = self._response_type(options=self.resp_options)

Expand Down Expand Up @@ -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()
Expand All @@ -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(
Expand Down Expand Up @@ -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:
Expand Down
Loading
Loading