fastapi - 💡(How to fix) Fix SSE `stream_item_type` not propagated through `APIRouter` + `include_router` [1 participants]

from collections.abc import AsyncIterator

from fastapi import APIRouter, FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel


class Frame(BaseModel):
    kind: str


# Case A — route registered directly on the app: works
app_a = FastAPI()

@app_a.post("/s", response_class=EventSourceResponse)
async def a() -> AsyncIterator[Frame]:
    yield Frame(kind="x")


# Case B — route defined on a router, then `include_router`: broken
router = APIRouter()

@router.post("/s", response_class=EventSourceResponse)
async def b() -> AsyncIterator[Frame]:
    yield Frame(kind="x")


app_b = FastAPI()
app_b.include_router(router)

print("DIRECT:", app_a.routes[-1].stream_item_type)
print("ROUTER (pre-include):", router.routes[-1].stream_item_type)
print("APP (post-include):", app_b.routes[-1].stream_item_type)


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", {})


print("DIRECT openapi has contentSchema:", has_content_schema(app_a.openapi()))
print("INCLUDE_ROUTER openapi has contentSchema:", has_content_schema(app_b.openapi()))

---

DIRECT: <class '__main__.Frame'>
ROUTER (pre-include): <class '__main__.Frame'>
APP (post-include): None

DIRECT openapi has contentSchema: True
INCLUDE_ROUTER openapi has contentSchema: False

---

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.add_api_route(
    prefix + route.path,
    route.endpoint,
    response_model=route.response_model,  # now None, not DefaultPlaceholder
    ...
    response_class=use_response_class,
    ...
)