From eca7bc9602b25658172a31a53640144bad7fc8e2 Mon Sep 17 00:00:00 2001 From: Jean-Baptiste Braun Date: Sat, 9 May 2026 16:30:32 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Propagate=20`stream=5Fitem=5Ftyp?= =?UTF-8?q?e`=20through=20`include=5Frouter`=20for=20SSE=20/=20JSONL=20rou?= =?UTF-8?q?tes=20(#15401)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When an SSE route (or JSONL streaming route) is defined on an `APIRouter` and merged onto a `FastAPI` app via `include_router`, the merged route silently lost its `stream_item_type`. As a consequence the emitted OpenAPI schema dropped the `contentSchema` describing the streamed item and downstream tools (`datamodel-codegen`, etc.) could no longer generate frame models from the spec. Root cause: `APIRoute.__init__` only ran stream-item detection inside the `isinstance(response_model, DefaultPlaceholder)` branch. The source route's `__init__` collapses `self.response_model` to `None` after detection, so when `APIRouter.include_router` re-instantiates the route via `add_api_route(response_model=route.response_model, ...)` the new init sees an explicit `None` and skips detection entirely. Fix: also run detection when `response_model is None` on entry, and restrict the "promote return annotation to response_model" fallback to the original `DefaultPlaceholder` case so an explicit `response_model= None` still disables response validation as documented. Adds a regression test asserting both `stream_item_type` propagation and the presence of `contentSchema` in the merged route's OpenAPI. Co-Authored-By: Claude Code --- fastapi/routing.py | 19 ++++++++++++++-- tests/test_sse.py | 57 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 2 deletions(-) diff --git a/fastapi/routing.py b/fastapi/routing.py index 36acb6b89d..d69da8be7b 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -844,7 +844,17 @@ class APIRoute(routing.Route): self.path = path self.endpoint = endpoint self.stream_item_type: Any | None = None - if isinstance(response_model, DefaultPlaceholder): + # Run stream-item detection both when response_model is the + # DefaultPlaceholder (the original code path: user-defined route) + # and when it's an explicit ``None``. The ``None`` case fires when + # ``include_router`` re-instantiates a route from a source whose + # ``response_model`` was already collapsed to ``None`` during the + # first ``__init__``: without it, the merged route would silently + # drop ``stream_item_type`` and the OpenAPI ``contentSchema`` for + # the streaming endpoint (#15401). Explicit ``response_model=None`` + # for streaming endpoints is also a documented pattern, so opting + # those into detection is intentional. + if isinstance(response_model, DefaultPlaceholder) or response_model is None: return_annotation = get_typed_return_annotation(endpoint) if lenient_issubclass(return_annotation, Response): response_model = None @@ -863,7 +873,12 @@ class APIRoute(routing.Route): ) and not lenient_issubclass(stream_item, ServerSentEvent): self.stream_item_type = stream_item response_model = None - else: + elif isinstance(response_model, DefaultPlaceholder): + # Only fall back to using the return annotation as the + # response_model when the caller did not pass an + # explicit value: an explicit ``None`` must stay + # ``None`` (otherwise we'd reintroduce response + # validation that the caller deliberately turned off). response_model = return_annotation self.response_model = response_model self.summary = summary diff --git a/tests/test_sse.py b/tests/test_sse.py index 6dfec61838..80bbaa9d81 100644 --- a/tests/test_sse.py +++ b/tests/test_sse.py @@ -316,3 +316,60 @@ def test_no_keepalive_when_fast(client: TestClient): assert response.status_code == 200 # KEEPALIVE_COMMENT is ": ping\n\n". assert ": ping\n" not in response.text + + +def test_stream_item_type_propagated_through_include_router(): + """Regression test for #15401. + + When an SSE route is defined on an ``APIRouter`` and merged onto a + ``FastAPI`` app via ``include_router``, the merged route must carry + the same ``stream_item_type`` as the source route, and the emitted + OpenAPI schema must include the ``contentSchema`` describing the + streamed item under + ``responses.200.content["text/event-stream"].itemSchema.properties.data``. + + Before the fix, ``stream_item_type`` detection in ``APIRoute.__init__`` + was gated on ``response_model`` being a ``DefaultPlaceholder``. The + source route's ``response_model`` was collapsed to ``None`` during + its first ``__init__``, so when ``include_router`` re-instantiated + the route with that ``None``, detection was skipped and the merged + route's ``stream_item_type`` stayed ``None``. + """ + + class Frame(BaseModel): + kind: str + + # Case A — route registered directly on the app (control) + app_a = FastAPI() + + @app_a.post("/s", response_class=EventSourceResponse) + async def direct() -> AsyncIterable[Frame]: + yield Frame(kind="x") + + # Case B — route on a router, then ``include_router`` (regression case) + router_b = APIRouter() + + @router_b.post("/s", response_class=EventSourceResponse) + async def via_router() -> AsyncIterable[Frame]: + yield Frame(kind="x") + + app_b = FastAPI() + app_b.include_router(router_b) + + # Both routes must carry the detected item type. + direct_route = app_a.routes[-1] + merged_route = app_b.routes[-1] + assert direct_route.stream_item_type is Frame # type: ignore[union-attr] + assert merged_route.stream_item_type is Frame # type: ignore[union-attr] + + # And both must surface the contentSchema in the emitted OpenAPI. + def has_content_schema(spec: dict) -> bool: + sse = spec["paths"]["/s"]["post"]["responses"]["200"]["content"][ + "text/event-stream" + ] + return "contentSchema" in sse.get("itemSchema", {}).get("properties", {}).get( + "data", {} + ) + + assert has_content_schema(app_a.openapi()) + assert has_content_schema(app_b.openapi())