Browse Source

Fix routing type-check failures in pre-commit.

Normalize request handler typing in routing utils and explicitly type APIRoute.response_field to satisfy mypy and ty hooks.

Made-with: Cursor
pull/15210/head
rajratanm 4 months ago
parent
commit
3c1dd7fdf2
  1. 3
      fastapi/routing_routes.py
  2. 14
      fastapi/routing_utils.py

3
fastapi/routing_routes.py

@ -163,6 +163,7 @@ class APIRoute(routing.Route):
if isinstance(status_code, IntEnum): if isinstance(status_code, IntEnum):
status_code = int(status_code) status_code = int(status_code)
self.status_code = status_code self.status_code = status_code
self.response_field: ModelField | None
if self.response_model: if self.response_model:
assert is_body_allowed_for_status_code(status_code), ( assert is_body_allowed_for_status_code(status_code), (
f"Status code {status_code} must not have a response body" f"Status code {status_code} must not have a response body"
@ -174,7 +175,7 @@ class APIRoute(routing.Route):
mode="serialization", mode="serialization",
) )
else: else:
self.response_field = None # type: ignore self.response_field = None
if self.stream_item_type: if self.stream_item_type:
stream_item_name = "StreamItem_" + self.unique_id stream_item_name = "StreamItem_" + self.unique_id
self.stream_item_field: ModelField | None = create_model_field( self.stream_item_field: ModelField | None = create_model_field(

14
fastapi/routing_utils.py

@ -15,7 +15,7 @@ from contextlib import (
AsyncExitStack, AsyncExitStack,
asynccontextmanager, asynccontextmanager,
) )
from typing import Any, TypeVar from typing import Any, TypeVar, cast
from fastapi._compat import ModelField from fastapi._compat import ModelField
from fastapi.dependencies.models import Dependant from fastapi.dependencies.models import Dependant
@ -40,9 +40,15 @@ _T = TypeVar("_T")
def request_response( def request_response(
func: Callable[[Request], Awaitable[Response] | Response], func: Callable[[Request], Awaitable[Response] | Response],
) -> ASGIApp: ) -> ASGIApp:
f: Callable[[Request], Awaitable[Response]] = ( if is_async_callable(func):
func if is_async_callable(func) else functools.partial(run_in_threadpool, func) # type: ignore[call-arg] f: Callable[[Request], Awaitable[Response]] = cast(
) Callable[[Request], Awaitable[Response]], func
)
else:
sync_func = cast(Callable[[Request], Response], func)
async def f(request: Request) -> Response:
return await run_in_threadpool(sync_func, request)
async def app(scope: Scope, receive: Receive, send: Send) -> None: async def app(scope: Scope, receive: Receive, send: Send) -> None:
request = Request(scope, receive, send) request = Request(scope, receive, send)

Loading…
Cancel
Save