Refactor routing internals into modular components with stable facade.

Extract APIRouter, APIRoute/APIWebSocketRoute, and handler/util logic into focused modules while preserving `fastapi.routing` public imports via compatibility re-exports and adding routing facade contract tests.

Made-with: Cursor

docs/en/docs/advanced/routing-architecture.md

@@ -0,0 +1,27 @@
+# Routing Architecture
+
+FastAPI keeps public routing imports stable at `fastapi.routing`, but the
+implementation is split into focused internal modules for maintainability.
+
+## Public facade
+
+- `fastapi.routing`
+  - compatibility facade and stable imports
+  - re-exports core symbols such as `APIRouter`, `APIRoute`, and helper hooks
+
+## Internal modules
+
+- `fastapi.routing_router`
+  - `APIRouter` implementation
+- `fastapi.routing_routes`
+  - `APIRoute` and `APIWebSocketRoute`
+- `fastapi.routing_handlers`
+  - request/websocket handler factories and handler config objects
+- `fastapi.routing_utils`
+  - low-level shared utilities (response serialization, lifespan helpers, wrappers)
+
+## Stability contract
+
+- Application code should import from `fastapi.routing` (or top-level `fastapi`).
+- Internal modules are implementation details and may evolve.
+- Compatibility tests validate that key re-exports remain stable.

fastapi/routing_handlers.py

@@ -0,0 +1,464 @@
+import email.message
+import json
+from collections.abc import AsyncIterator, Callable, Coroutine, Iterator
+from contextlib import AsyncExitStack, asynccontextmanager
+from dataclasses import dataclass, field
+from typing import Any, cast
+
+import anyio
+from anyio.abc import ObjectReceiveStream
+from fastapi import params
+from fastapi._compat import ModelField, Undefined, lenient_issubclass
+from fastapi.datastructures import Default, DefaultPlaceholder
+from fastapi.dependencies.models import Dependant
+from fastapi.dependencies.utils import solve_dependencies
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import (
+    EndpointContext,
+    RequestValidationError,
+    ResponseValidationError,
+    WebSocketRequestValidationError,
+)
+from fastapi.routing_utils import (
+    build_response_args,
+    extract_endpoint_context,
+    run_endpoint_function,
+    serialize_response,
+)
+from fastapi.sse import (
+    KEEPALIVE_COMMENT,
+    EventSourceResponse,
+    ServerSentEvent,
+    format_sse_event,
+)
+from fastapi.types import IncEx
+from fastapi.utils import is_body_allowed_for_status_code
+from starlette.concurrency import iterate_in_threadpool
+from starlette.datastructures import FormData
+from starlette.exceptions import HTTPException
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response, StreamingResponse
+from starlette.websockets import WebSocket
+
+
+def _get_ping_interval() -> float:
+    # Keep compatibility with tests/users that patch fastapi.routing._PING_INTERVAL.
+    # If not present, fall back to the default defined in fastapi.sse.
+    from fastapi import routing as routing_module
+    from fastapi.sse import _PING_INTERVAL as default_ping_interval
+
+    return float(getattr(routing_module, "_PING_INTERVAL", default_ping_interval))
+
+
+@dataclass
+class RouteHandlerConfig:
+    dependant: Dependant
+    body_field: ModelField | None = None
+    status_code: int | None = None
+    response_class: type[Response] | DefaultPlaceholder = field(
+        default_factory=lambda: Default(JSONResponse)
+    )
+    response_field: ModelField | None = None
+    response_model_include: IncEx | None = None
+    response_model_exclude: IncEx | None = None
+    response_model_by_alias: bool = True
+    response_model_exclude_unset: bool = False
+    response_model_exclude_defaults: bool = False
+    response_model_exclude_none: bool = False
+    dependency_overrides_provider: Any | None = None
+    embed_body_fields: bool = False
+    strict_content_type: bool | DefaultPlaceholder = field(
+        default_factory=lambda: Default(True)
+    )
+    stream_item_field: ModelField | None = None
+    is_json_stream: bool = False
+
+
+def get_request_handler(
+    dependant: Dependant | RouteHandlerConfig,
+    body_field: ModelField | None = None,
+    status_code: int | None = None,
+    response_class: type[Response] | DefaultPlaceholder = Default(JSONResponse),
+    response_field: ModelField | None = None,
+    response_model_include: IncEx | None = None,
+    response_model_exclude: IncEx | None = None,
+    response_model_by_alias: bool = True,
+    response_model_exclude_unset: bool = False,
+    response_model_exclude_defaults: bool = False,
+    response_model_exclude_none: bool = False,
+    dependency_overrides_provider: Any | None = None,
+    embed_body_fields: bool = False,
+    strict_content_type: bool | DefaultPlaceholder = Default(True),
+    stream_item_field: ModelField | None = None,
+    is_json_stream: bool = False,
+) -> Callable[[Request], Coroutine[Any, Any, Response]]:
+    if isinstance(dependant, RouteHandlerConfig):
+        config = dependant
+    else:
+        config = RouteHandlerConfig(
+            dependant=dependant,
+            body_field=body_field,
+            status_code=status_code,
+            response_class=response_class,
+            response_field=response_field,
+            response_model_include=response_model_include,
+            response_model_exclude=response_model_exclude,
+            response_model_by_alias=response_model_by_alias,
+            response_model_exclude_unset=response_model_exclude_unset,
+            response_model_exclude_defaults=response_model_exclude_defaults,
+            response_model_exclude_none=response_model_exclude_none,
+            dependency_overrides_provider=dependency_overrides_provider,
+            embed_body_fields=embed_body_fields,
+            strict_content_type=strict_content_type,
+            stream_item_field=stream_item_field,
+            is_json_stream=is_json_stream,
+        )
+    dependant = config.dependant
+    body_field = config.body_field
+    status_code = config.status_code
+    response_class = config.response_class
+    response_field = config.response_field
+    response_model_include = config.response_model_include
+    response_model_exclude = config.response_model_exclude
+    response_model_by_alias = config.response_model_by_alias
+    response_model_exclude_unset = config.response_model_exclude_unset
+    response_model_exclude_defaults = config.response_model_exclude_defaults
+    response_model_exclude_none = config.response_model_exclude_none
+    dependency_overrides_provider = config.dependency_overrides_provider
+    embed_body_fields = config.embed_body_fields
+    strict_content_type = config.strict_content_type
+    stream_item_field = config.stream_item_field
+    is_json_stream = config.is_json_stream
+
+    assert dependant.call is not None, "dependant.call must be a function"
+    is_coroutine = dependant.is_coroutine_callable
+    is_body_form = body_field and isinstance(body_field.field_info, params.Form)
+    if isinstance(response_class, DefaultPlaceholder):
+        actual_response_class: type[Response] = response_class.value
+    else:
+        actual_response_class = response_class
+    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)
+    if isinstance(strict_content_type, DefaultPlaceholder):
+        actual_strict_content_type: bool = strict_content_type.value
+    else:
+        actual_strict_content_type = strict_content_type
+
+    async def app(request: Request) -> Response:
+        response: Response | None = None
+        file_stack = request.scope.get("fastapi_middleware_astack")
+        assert isinstance(file_stack, AsyncExitStack), (
+            "fastapi_middleware_astack not found in request scope"
+        )
+
+        endpoint_ctx = (
+            extract_endpoint_context(dependant.call)
+            if dependant.call
+            else EndpointContext()
+        )
+
+        if dependant.path:
+            mount_path = request.scope.get("root_path", "").rstrip("/")
+            endpoint_ctx["path"] = f"{request.method} {mount_path}{dependant.path}"
+
+        try:
+            body: Any = None
+            if body_field:
+                if is_body_form:
+                    body = await request.form()
+                    file_stack.push_async_callback(body.close)
+                else:
+                    body_bytes = await request.body()
+                    if body_bytes:
+                        json_body: Any = Undefined
+                        content_type_value = request.headers.get("content-type")
+                        if not content_type_value:
+                            if not actual_strict_content_type:
+                                json_body = await request.json()
+                        else:
+                            message = email.message.Message()
+                            message["content-type"] = content_type_value
+                            if message.get_content_maintype() == "application":
+                                subtype = message.get_content_subtype()
+                                if subtype == "json" or subtype.endswith("+json"):
+                                    json_body = await request.json()
+                        if json_body != Undefined:
+                            body = json_body
+                        else:
+                            body = body_bytes
+        except json.JSONDecodeError as e:
+            validation_error = RequestValidationError(
+                [
+                    {
+                        "type": "json_invalid",
+                        "loc": ("body", e.pos),
+                        "msg": "JSON decode error",
+                        "input": {},
+                        "ctx": {"error": e.msg},
+                    }
+                ],
+                body=e.doc,
+                endpoint_ctx=endpoint_ctx,
+            )
+            raise validation_error from e
+        except HTTPException:
+            raise
+        except Exception as e:
+            http_error = HTTPException(
+                status_code=400, detail="There was an error parsing the body"
+            )
+            raise http_error from e
+
+        async_exit_stack = request.scope.get("fastapi_inner_astack")
+        assert isinstance(async_exit_stack, AsyncExitStack), (
+            "fastapi_inner_astack not found in request scope"
+        )
+        solved_result = await solve_dependencies(
+            request=request,
+            dependant=dependant,
+            body=cast(dict[str, Any] | FormData | bytes | None, body),
+            dependency_overrides_provider=dependency_overrides_provider,
+            async_exit_stack=async_exit_stack,
+            embed_body_fields=embed_body_fields,
+        )
+        errors = solved_result.errors
+        assert dependant.call
+        if not errors:
+            def _serialize_data(data: Any) -> bytes:
+                if stream_item_field:
+                    value, errors_ = stream_item_field.validate(data, {}, loc=("response",))
+                    if errors_:
+                        ctx = endpoint_ctx or EndpointContext()
+                        raise ResponseValidationError(
+                            errors=errors_,
+                            body=data,
+                            endpoint_ctx=ctx,
+                        )
+                    return stream_item_field.serialize_json(
+                        value,
+                        include=response_model_include,
+                        exclude=response_model_exclude,
+                        by_alias=response_model_by_alias,
+                        exclude_unset=response_model_exclude_unset,
+                        exclude_defaults=response_model_exclude_defaults,
+                        exclude_none=response_model_exclude_none,
+                    )
+                data = jsonable_encoder(data)
+                return json.dumps(data).encode("utf-8")
+
+            if is_sse_stream:
+                gen = dependant.call(**solved_result.values)
+
+                def _serialize_sse_item(item: Any) -> bytes:
+                    if isinstance(item, ServerSentEvent):
+                        if item.raw_data is not None:
+                            data_str: str | None = item.raw_data
+                        elif item.data is not None:
+                            if hasattr(item.data, "model_dump_json"):
+                                data_str = item.data.model_dump_json()
+                            else:
+                                data_str = json.dumps(jsonable_encoder(item.data))
+                        else:
+                            data_str = None
+                        return format_sse_event(
+                            data_str=data_str,
+                            event=item.event,
+                            id=item.id,
+                            retry=item.retry,
+                            comment=item.comment,
+                        )
+                    return format_sse_event(data_str=_serialize_data(item).decode("utf-8"))
+
+                if dependant.is_async_gen_callable:
+                    sse_aiter: AsyncIterator[Any] = gen.__aiter__()
+                else:
+                    sse_aiter = iterate_in_threadpool(gen)
+
+                @asynccontextmanager
+                async def _sse_producer_cm() -> AsyncIterator[ObjectReceiveStream[bytes]]:
+                    send_stream, receive_stream = anyio.create_memory_object_stream[bytes](
+                        max_buffer_size=1
+                    )
+
+                    async def _producer() -> None:
+                        async with send_stream:
+                            async for raw_item in sse_aiter:
+                                await send_stream.send(_serialize_sse_item(raw_item))
+
+                    send_keepalive, receive_keepalive = anyio.create_memory_object_stream[
+                        bytes
+                    ](max_buffer_size=1)
+
+                    async def _keepalive_inserter() -> None:
+                        async with send_keepalive, receive_stream:
+                            try:
+                                while True:
+                                    try:
+                                        with anyio.fail_after(_get_ping_interval()):
+                                            data = await receive_stream.receive()
+                                        await send_keepalive.send(data)
+                                    except TimeoutError:
+                                        await send_keepalive.send(KEEPALIVE_COMMENT)
+                            except anyio.EndOfStream:
+                                pass
+
+                    async with anyio.create_task_group() as tg:
+                        tg.start_soon(_producer)
+                        tg.start_soon(_keepalive_inserter)
+                        yield receive_keepalive
+                        tg.cancel_scope.cancel()
+
+                sse_receive_stream = await async_exit_stack.enter_async_context(
+                    _sse_producer_cm()
+                )
+                async_exit_stack.push_async_callback(sse_receive_stream.aclose)
+
+                async def _sse_with_checkpoints(
+                    stream: ObjectReceiveStream[bytes],
+                ) -> AsyncIterator[bytes]:
+                    async for data in stream:
+                        yield data
+                        await anyio.sleep(0)
+
+                sse_stream_content: AsyncIterator[bytes] | Iterator[bytes] = (
+                    _sse_with_checkpoints(sse_receive_stream)
+                )
+
+                response = StreamingResponse(
+                    sse_stream_content,
+                    media_type="text/event-stream",
+                    background=solved_result.background_tasks,
+                )
+                response.headers["Cache-Control"] = "no-cache"
+                response.headers["X-Accel-Buffering"] = "no"
+                response.headers.raw.extend(solved_result.response.headers.raw)
+            elif is_json_stream:
+                gen = dependant.call(**solved_result.values)
+
+                def _serialize_item(item: Any) -> bytes:
+                    return _serialize_data(item) + b"\n"
+
+                if dependant.is_async_gen_callable:
+
+                    async def _async_stream_jsonl() -> AsyncIterator[bytes]:
+                        async for item in gen:
+                            yield _serialize_item(item)
+                            await anyio.sleep(0)
+
+                    jsonl_stream_content: AsyncIterator[bytes] | Iterator[bytes] = (
+                        _async_stream_jsonl()
+                    )
+                else:
+
+                    def _sync_stream_jsonl() -> Iterator[bytes]:
+                        for item in gen:  # ty: ignore[not-iterable]
+                            yield _serialize_item(item)
+
+                    jsonl_stream_content = _sync_stream_jsonl()
+
+                response = StreamingResponse(
+                    jsonl_stream_content,
+                    media_type="application/jsonl",
+                    background=solved_result.background_tasks,
+                )
+                response.headers.raw.extend(solved_result.response.headers.raw)
+            elif dependant.is_async_gen_callable or dependant.is_gen_callable:
+                gen = dependant.call(**solved_result.values)
+                if dependant.is_async_gen_callable:
+
+                    async def _async_stream_raw(
+                        async_gen: AsyncIterator[Any],
+                    ) -> AsyncIterator[Any]:
+                        async for chunk in async_gen:
+                            yield chunk
+                            await anyio.sleep(0)
+
+                    gen = _async_stream_raw(gen)
+                response_args = build_response_args(
+                    status_code=status_code, solved_result=solved_result
+                )
+                response = actual_response_class(content=gen, **response_args)
+                response.headers.raw.extend(solved_result.response.headers.raw)
+            else:
+                raw_response = await run_endpoint_function(
+                    dependant=dependant,
+                    values=solved_result.values,
+                    is_coroutine=is_coroutine,
+                )
+                if isinstance(raw_response, Response):
+                    if raw_response.background is None:
+                        raw_response.background = solved_result.background_tasks
+                    response = raw_response
+                else:
+                    response_args = build_response_args(
+                        status_code=status_code, solved_result=solved_result
+                    )
+                    use_dump_json = response_field is not None and isinstance(
+                        response_class, DefaultPlaceholder
+                    )
+                    content = await serialize_response(
+                        field=response_field,
+                        response_content=raw_response,
+                        include=response_model_include,
+                        exclude=response_model_exclude,
+                        by_alias=response_model_by_alias,
+                        exclude_unset=response_model_exclude_unset,
+                        exclude_defaults=response_model_exclude_defaults,
+                        exclude_none=response_model_exclude_none,
+                        is_coroutine=is_coroutine,
+                        endpoint_ctx=endpoint_ctx,
+                        dump_json=use_dump_json,
+                    )
+                    if use_dump_json:
+                        response = Response(
+                            content=content,
+                            media_type="application/json",
+                            **response_args,
+                        )
+                    else:
+                        response = actual_response_class(content, **response_args)
+                    if not is_body_allowed_for_status_code(response.status_code):
+                        response.body = b""
+                    response.headers.raw.extend(solved_result.response.headers.raw)
+        if errors:
+            raise RequestValidationError(errors, body=body, endpoint_ctx=endpoint_ctx)
+
+        assert response
+        return response
+
+    return app
+
+
+def get_websocket_app(
+    dependant: Dependant,
+    dependency_overrides_provider: Any | None = None,
+    embed_body_fields: bool = False,
+) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:
+    async def app(websocket: WebSocket) -> None:
+        endpoint_ctx = (
+            extract_endpoint_context(dependant.call)
+            if dependant.call
+            else EndpointContext()
+        )
+        if dependant.path:
+            mount_path = websocket.scope.get("root_path", "").rstrip("/")
+            endpoint_ctx["path"] = f"WS {mount_path}{dependant.path}"
+        async_exit_stack = websocket.scope.get("fastapi_inner_astack")
+        assert isinstance(async_exit_stack, AsyncExitStack), (
+            "fastapi_inner_astack not found in request scope"
+        )
+        solved_result = await solve_dependencies(
+            request=websocket,
+            dependant=dependant,
+            dependency_overrides_provider=dependency_overrides_provider,
+            async_exit_stack=async_exit_stack,
+            embed_body_fields=embed_body_fields,
+        )
+        if solved_result.errors:
+            raise WebSocketRequestValidationError(
+                solved_result.errors,
+                endpoint_ctx=endpoint_ctx,
+            )
+        assert dependant.call is not None, "dependant.call must be a function"
+        await dependant.call(**solved_result.values)
+
+    return app

fastapi/routing_routes.py

@@ -0,0 +1,259 @@
+import inspect
+from collections.abc import Callable, Coroutine, Sequence
+from enum import Enum, IntEnum
+from typing import Any
+
+from fastapi import params
+from fastapi._compat import ModelField, lenient_issubclass
+from fastapi.datastructures import Default, DefaultPlaceholder
+from fastapi.dependencies.utils import (
+    _should_embed_body_fields,
+    get_body_field,
+    get_dependant,
+    get_flat_dependant,
+    get_parameterless_sub_dependant,
+    get_stream_item_type,
+    get_typed_return_annotation,
+)
+from fastapi.routing_handlers import (
+    RouteHandlerConfig,
+    get_request_handler,
+    get_websocket_app,
+)
+from fastapi.routing_utils import request_response, websocket_session
+from fastapi.sse import EventSourceResponse, ServerSentEvent
+from fastapi.types import IncEx
+from fastapi.utils import create_model_field, generate_unique_id, is_body_allowed_for_status_code
+from starlette import routing
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import BaseRoute, Match, compile_path, get_name
+from starlette.types import Scope
+
+
+class APIWebSocketRoute(routing.WebSocketRoute):
+    def __init__(
+        self,
+        path: str,
+        endpoint: Callable[..., Any],
+        *,
+        name: str | None = None,
+        dependencies: Sequence[params.Depends] | None = None,
+        dependency_overrides_provider: Any | None = None,
+    ) -> None:
+        self.path = path
+        self.endpoint = endpoint
+        self.name = get_name(endpoint) if name is None else name
+        self.dependencies = list(dependencies or [])
+        self.path_regex, self.path_format, self.param_convertors = compile_path(path)
+        self.dependant = get_dependant(
+            path=self.path_format, call=self.endpoint, scope="function"
+        )
+        for depends in self.dependencies[::-1]:
+            self.dependant.dependencies.insert(
+                0,
+                get_parameterless_sub_dependant(depends=depends, path=self.path_format),
+            )
+        self._flat_dependant = get_flat_dependant(self.dependant)
+        self._embed_body_fields = _should_embed_body_fields(
+            self._flat_dependant.body_params
+        )
+        self.app = websocket_session(
+            get_websocket_app(
+                dependant=self.dependant,
+                dependency_overrides_provider=dependency_overrides_provider,
+                embed_body_fields=self._embed_body_fields,
+            )
+        )
+
+    def matches(self, scope: Scope) -> tuple[Match, Scope]:
+        match, child_scope = super().matches(scope)
+        if match != Match.NONE:
+            child_scope["route"] = self
+        return match, child_scope
+
+
+class APIRoute(routing.Route):
+    def __init__(
+        self,
+        path: str,
+        endpoint: Callable[..., Any],
+        *,
+        response_model: Any = Default(None),
+        status_code: int | None = None,
+        tags: list[str | Enum] | None = None,
+        dependencies: Sequence[params.Depends] | None = None,
+        summary: str | None = None,
+        description: str | None = None,
+        response_description: str = "Successful Response",
+        responses: dict[int | str, dict[str, Any]] | None = None,
+        deprecated: bool | None = None,
+        name: str | None = None,
+        methods: set[str] | list[str] | None = None,
+        operation_id: str | None = None,
+        response_model_include: IncEx | None = None,
+        response_model_exclude: IncEx | None = None,
+        response_model_by_alias: bool = True,
+        response_model_exclude_unset: bool = False,
+        response_model_exclude_defaults: bool = False,
+        response_model_exclude_none: bool = False,
+        include_in_schema: bool = True,
+        response_class: type[Response] | DefaultPlaceholder = Default(JSONResponse),
+        dependency_overrides_provider: Any | None = None,
+        callbacks: list[BaseRoute] | None = None,
+        openapi_extra: dict[str, Any] | None = None,
+        generate_unique_id_function: Callable[["APIRoute"], str]
+        | DefaultPlaceholder = Default(generate_unique_id),
+        strict_content_type: bool | DefaultPlaceholder = Default(True),
+    ) -> None:
+        self.path = path
+        self.endpoint = endpoint
+        self.stream_item_type: Any | None = None
+        if isinstance(response_model, DefaultPlaceholder):
+            return_annotation = get_typed_return_annotation(endpoint)
+            if lenient_issubclass(return_annotation, Response):
+                response_model = None
+            else:
+                stream_item = get_stream_item_type(return_annotation)
+                if stream_item is not None:
+                    if (
+                        isinstance(response_class, DefaultPlaceholder)
+                        or lenient_issubclass(response_class, EventSourceResponse)
+                    ) and not lenient_issubclass(stream_item, ServerSentEvent):
+                        self.stream_item_type = stream_item
+                    response_model = None
+                else:
+                    response_model = return_annotation
+        self.response_model = response_model
+        self.summary = summary
+        self.response_description = response_description
+        self.deprecated = deprecated
+        self.operation_id = operation_id
+        self.response_model_include = response_model_include
+        self.response_model_exclude = response_model_exclude
+        self.response_model_by_alias = response_model_by_alias
+        self.response_model_exclude_unset = response_model_exclude_unset
+        self.response_model_exclude_defaults = response_model_exclude_defaults
+        self.response_model_exclude_none = response_model_exclude_none
+        self.include_in_schema = include_in_schema
+        self.response_class = response_class
+        self.dependency_overrides_provider = dependency_overrides_provider
+        self.callbacks = callbacks
+        self.openapi_extra = openapi_extra
+        self.generate_unique_id_function = generate_unique_id_function
+        self.strict_content_type = strict_content_type
+        self.tags = tags or []
+        self.responses = responses or {}
+        self.name = get_name(endpoint) if name is None else name
+        self.path_regex, self.path_format, self.param_convertors = compile_path(path)
+        if methods is None:
+            methods = ["GET"]
+        self.methods: set[str] = {method.upper() for method in methods}
+        if isinstance(generate_unique_id_function, DefaultPlaceholder):
+            current_generate_unique_id: Callable[[APIRoute], str] = (
+                generate_unique_id_function.value
+            )
+        else:
+            current_generate_unique_id = generate_unique_id_function
+        self.unique_id = self.operation_id or current_generate_unique_id(self)
+        if isinstance(status_code, IntEnum):
+            status_code = int(status_code)
+        self.status_code = status_code
+        if self.response_model:
+            assert is_body_allowed_for_status_code(status_code), (
+                f"Status code {status_code} must not have a response body"
+            )
+            response_name = "Response_" + self.unique_id
+            self.response_field = create_model_field(
+                name=response_name,
+                type_=self.response_model,
+                mode="serialization",
+            )
+        else:
+            self.response_field = None  # type: ignore
+        if self.stream_item_type:
+            stream_item_name = "StreamItem_" + self.unique_id
+            self.stream_item_field: ModelField | None = create_model_field(
+                name=stream_item_name,
+                type_=self.stream_item_type,
+                mode="serialization",
+            )
+        else:
+            self.stream_item_field = None
+        self.dependencies = list(dependencies or [])
+        self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "")
+        self.description = self.description.split("\f")[0].strip()
+        response_fields = {}
+        for additional_status_code, response in self.responses.items():
+            assert isinstance(response, dict), "An additional response must be a dict"
+            model = response.get("model")
+            if model:
+                assert is_body_allowed_for_status_code(additional_status_code), (
+                    f"Status code {additional_status_code} must not have a response body"
+                )
+                response_name = f"Response_{additional_status_code}_{self.unique_id}"
+                response_field = create_model_field(
+                    name=response_name, type_=model, mode="serialization"
+                )
+                response_fields[additional_status_code] = response_field
+        if response_fields:
+            self.response_fields: dict[int | str, ModelField] = response_fields
+        else:
+            self.response_fields = {}
+
+        assert callable(endpoint), "An endpoint must be a callable"
+        self.dependant = get_dependant(
+            path=self.path_format, call=self.endpoint, scope="function"
+        )
+        for depends in self.dependencies[::-1]:
+            self.dependant.dependencies.insert(
+                0,
+                get_parameterless_sub_dependant(depends=depends, path=self.path_format),
+            )
+        self._flat_dependant = get_flat_dependant(self.dependant)
+        self._embed_body_fields = _should_embed_body_fields(
+            self._flat_dependant.body_params
+        )
+        self.body_field = get_body_field(
+            flat_dependant=self._flat_dependant,
+            name=self.unique_id,
+            embed_body_fields=self._embed_body_fields,
+        )
+        is_generator = (
+            self.dependant.is_async_gen_callable or self.dependant.is_gen_callable
+        )
+        self.is_sse_stream = is_generator and lenient_issubclass(
+            response_class, EventSourceResponse
+        )
+        self.is_json_stream = is_generator and isinstance(
+            response_class, DefaultPlaceholder
+        )
+        self.app = request_response(self.get_route_handler())
+
+    def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
+        return get_request_handler(
+            RouteHandlerConfig(
+            dependant=self.dependant,
+            body_field=self.body_field,
+            status_code=self.status_code,
+            response_class=self.response_class,
+            response_field=self.response_field,
+            response_model_include=self.response_model_include,
+            response_model_exclude=self.response_model_exclude,
+            response_model_by_alias=self.response_model_by_alias,
+            response_model_exclude_unset=self.response_model_exclude_unset,
+            response_model_exclude_defaults=self.response_model_exclude_defaults,
+            response_model_exclude_none=self.response_model_exclude_none,
+            dependency_overrides_provider=self.dependency_overrides_provider,
+            embed_body_fields=self._embed_body_fields,
+            strict_content_type=self.strict_content_type,
+            stream_item_field=self.stream_item_field,
+            is_json_stream=self.is_json_stream,
+            )
+        )
+
+    def matches(self, scope: Scope) -> tuple[Match, Scope]:
+        match, child_scope = super().matches(scope)
+        if match != Match.NONE:
+            child_scope["route"] = self
+        return match, child_scope

fastapi/routing_utils.py

@@ -0,0 +1,231 @@
+import contextlib
+import functools
+import inspect
+import types
+from collections.abc import (
+    AsyncIterator,
+    Awaitable,
+    Callable,
+    Coroutine,
+    Generator,
+    Mapping,
+)
+from contextlib import (
+    AbstractAsyncContextManager,
+    AbstractContextManager,
+    AsyncExitStack,
+    asynccontextmanager,
+)
+from typing import Any, TypeVar
+
+from fastapi._compat import ModelField
+from fastapi.dependencies.models import Dependant
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import (
+    EndpointContext,
+    FastAPIError,
+    ResponseValidationError,
+)
+from fastapi.types import IncEx
+from starlette._exception_handler import wrap_app_handling_exceptions
+from starlette._utils import is_async_callable
+from starlette.concurrency import run_in_threadpool
+from starlette.requests import Request
+from starlette.responses import Response
+from starlette.types import ASGIApp, AppType, Lifespan, Receive, Scope, Send
+from starlette.websockets import WebSocket
+
+_T = TypeVar("_T")
+
+
+def request_response(
+    func: Callable[[Request], Awaitable[Response] | Response],
+) -> ASGIApp:
+    f: Callable[[Request], Awaitable[Response]] = (
+        func
+        if is_async_callable(func)
+        else functools.partial(run_in_threadpool, func)  # type: ignore[call-arg]
+    )
+
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        request = Request(scope, receive, send)
+
+        async def app(scope: Scope, receive: Receive, send: Send) -> None:
+            response_awaited = False
+            async with AsyncExitStack() as request_stack:
+                scope["fastapi_inner_astack"] = request_stack
+                async with AsyncExitStack() as function_stack:
+                    scope["fastapi_function_astack"] = function_stack
+                    response = await f(request)
+                await response(scope, receive, send)
+                response_awaited = True
+            if not response_awaited:
+                raise FastAPIError(
+                    "Response not awaited. There's a high chance that the "
+                    "application code is raising an exception and a dependency with yield "
+                    "has a block with a bare except, or a block with except Exception, "
+                    "and is not raising the exception again. Read more about it in the "
+                    "docs: https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/#dependencies-with-yield-and-except"
+                )
+
+        await wrap_app_handling_exceptions(app, request)(scope, receive, send)
+
+    return app
+
+
+def websocket_session(
+    func: Callable[[WebSocket], Awaitable[None]],
+) -> ASGIApp:
+    async def app(scope: Scope, receive: Receive, send: Send) -> None:
+        session = WebSocket(scope, receive=receive, send=send)
+
+        async def app(scope: Scope, receive: Receive, send: Send) -> None:
+            async with AsyncExitStack() as request_stack:
+                scope["fastapi_inner_astack"] = request_stack
+                async with AsyncExitStack() as function_stack:
+                    scope["fastapi_function_astack"] = function_stack
+                    await func(session)
+
+        await wrap_app_handling_exceptions(app, session)(scope, receive, send)
+
+    return app
+
+
+class _AsyncLiftContextManager(AbstractAsyncContextManager[_T]):
+    def __init__(self, cm: AbstractContextManager[_T]) -> None:
+        self._cm = cm
+
+    async def __aenter__(self) -> _T:
+        return self._cm.__enter__()
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: types.TracebackType | None,
+    ) -> bool | None:
+        return self._cm.__exit__(exc_type, exc_value, traceback)
+
+
+def _wrap_gen_lifespan_context(
+    lifespan_context: Callable[[Any], Generator[Any, Any, Any]],
+) -> Callable[[Any], AbstractAsyncContextManager[Any]]:
+    cmgr = contextlib.contextmanager(lifespan_context)
+
+    @functools.wraps(cmgr)
+    def wrapper(app: Any) -> _AsyncLiftContextManager[Any]:
+        return _AsyncLiftContextManager(cmgr(app))
+
+    return wrapper
+
+
+def _merge_lifespan_context(
+    original_context: Lifespan[Any], nested_context: Lifespan[Any]
+) -> Lifespan[Any]:
+    @asynccontextmanager
+    async def merged_lifespan(
+        app: AppType,
+    ) -> AsyncIterator[Mapping[str, Any] | None]:
+        async with original_context(app) as maybe_original_state:
+            async with nested_context(app) as maybe_nested_state:
+                if maybe_nested_state is None and maybe_original_state is None:
+                    yield None
+                else:
+                    yield {**(maybe_nested_state or {}), **(maybe_original_state or {})}
+
+    return merged_lifespan  # type: ignore[return-value]
+
+
+class _DefaultLifespan:
+    def __init__(self, router: Any) -> None:
+        self._router = router
+
+    async def __aenter__(self) -> None:
+        await self._router._startup()
+
+    async def __aexit__(self, *exc_info: object) -> None:
+        await self._router._shutdown()
+
+    def __call__(self: _T, app: object) -> _T:
+        return self
+
+
+_endpoint_context_cache: dict[int, EndpointContext] = {}
+
+
+def extract_endpoint_context(func: Any) -> EndpointContext:
+    func_id = id(func)
+    if func_id in _endpoint_context_cache:
+        return _endpoint_context_cache[func_id]
+    try:
+        ctx: EndpointContext = {}
+        if (source_file := inspect.getsourcefile(func)) is not None:
+            ctx["file"] = source_file
+        if (line_number := inspect.getsourcelines(func)[1]) is not None:
+            ctx["line"] = line_number
+        if (func_name := getattr(func, "__name__", None)) is not None:
+            ctx["function"] = func_name
+    except Exception:
+        ctx = EndpointContext()
+    _endpoint_context_cache[func_id] = ctx
+    return ctx
+
+
+async def serialize_response(
+    *,
+    field: ModelField | None = None,
+    response_content: Any,
+    include: IncEx | None = None,
+    exclude: IncEx | None = None,
+    by_alias: bool = True,
+    exclude_unset: bool = False,
+    exclude_defaults: bool = False,
+    exclude_none: bool = False,
+    is_coroutine: bool = True,
+    endpoint_ctx: EndpointContext | None = None,
+    dump_json: bool = False,
+) -> Any:
+    if field:
+        if is_coroutine:
+            value, errors = field.validate(response_content, {}, loc=("response",))
+        else:
+            value, errors = await run_in_threadpool(
+                field.validate, response_content, {}, loc=("response",)
+            )
+        if errors:
+            ctx = endpoint_ctx or EndpointContext()
+            raise ResponseValidationError(
+                errors=errors,
+                body=response_content,
+                endpoint_ctx=ctx,
+            )
+        serializer = field.serialize_json if dump_json else field.serialize
+        return serializer(
+            value,
+            include=include,
+            exclude=exclude,
+            by_alias=by_alias,
+            exclude_unset=exclude_unset,
+            exclude_defaults=exclude_defaults,
+            exclude_none=exclude_none,
+        )
+    return jsonable_encoder(response_content)
+
+
+async def run_endpoint_function(
+    *, dependant: Dependant, values: dict[str, Any], is_coroutine: bool
+) -> Any:
+    assert dependant.call is not None, "dependant.call must be a function"
+    if is_coroutine:
+        return await dependant.call(**values)
+    return await run_in_threadpool(dependant.call, **values)
+
+
+def build_response_args(*, status_code: int | None, solved_result: Any) -> dict[str, Any]:
+    response_args: dict[str, Any] = {"background": solved_result.background_tasks}
+    current_status_code = status_code if status_code else solved_result.response.status_code
+    if current_status_code is not None:
+        response_args["status_code"] = current_status_code
+    if solved_result.response.status_code:
+        response_args["status_code"] = solved_result.response.status_code
+    return response_args

tests/test_routing_facade_contract.py

@@ -0,0 +1,37 @@
+import fastapi.routing as routing
+
+from fastapi.routing_handlers import RouteHandlerConfig
+from fastapi.routing_router import APIRouter
+from fastapi.routing_routes import APIRoute, APIWebSocketRoute
+from fastapi.routing_utils import request_response, websocket_session
+
+
+def test_routing_facade_exports_expected_symbols() -> None:
+    expected = {
+        "APIRouter",
+        "APIRoute",
+        "APIWebSocketRoute",
+        "RouteHandlerConfig",
+        "get_request_handler",
+        "get_websocket_app",
+        "request_response",
+        "websocket_session",
+        "_PING_INTERVAL",
+    }
+    assert expected.issubset(set(routing.__all__))
+
+
+def test_routing_facade_re_exports_router_classes() -> None:
+    assert routing.APIRouter is APIRouter
+    assert routing.APIRoute is APIRoute
+    assert routing.APIWebSocketRoute is APIWebSocketRoute
+
+
+def test_routing_facade_re_exports_handler_contracts() -> None:
+    assert routing.RouteHandlerConfig is RouteHandlerConfig
+    assert routing.request_response is request_response
+    assert routing.websocket_session is websocket_session
+
+
+def test_routing_ping_constant_compatibility() -> None:
+    assert isinstance(routing._PING_INTERVAL, (float, int))