✨ Make ServerSentEvent generic with typed data field#15191
✨ Make ServerSentEvent generic with typed data field#15191benmosher wants to merge 6 commits intofastapi:masterfrom
Conversation
Introduce `ServerSentEvent[Data]` so endpoints that yield typed SSE events get a `contentSchema` in the OpenAPI spec reflecting the data payload type, while retaining full control over SSE fields (`event`, `id`, `retry`, `comment`). - `ServerSentEvent` now inherits from `Generic[Data]` with `data: Data` - `validate_default=True` ensures `ServerSentEvent[Item]()` raises a ValidationError (data is effectively required when Data is concrete) - `ServerSentEvent[Item | None]` allows optional data; bare `ServerSentEvent` is fully backward compatible (`Data=Any`) - Added `get_sse_data_type()` helper (uses Pydantic's `__pydantic_generic_metadata__`) to extract `Data` from a parameterized `ServerSentEvent[Data]` annotation - Routing layer now extracts `Data` from `ServerSentEvent[Data]` and uses it as `stream_item_type`, feeding it into the existing OpenAPI `contentSchema` pipeline - Added tutorial006 and corresponding snapshot test - Extended test_sse.py with generic SSE unit and app-level tests Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
The previous commit added ServerSentEvent[Data] generic support and tutorial006 but did not add a corresponding section to the SSE docs. This adds the missing "Typed ServerSentEvent" section explaining the generic syntax, validation behavior, and OpenAPI contentSchema benefit. https://claude.ai/code/session_01GRv2sCmACvBQFX7fh7anWm
📝 Docs previewLast commit f6d54bf at: https://ad2a27b2.fastapitiangolo.pages.dev Modified Pages |
Bare `ServerSentEvent` (without a type parameter) now silently resolves to `ServerSentEvent[Any]`, so existing code and the `_check_data_exclusive` return annotation require no changes and pass mypy without `[type-arg]` errors. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…e test Pydantic BaseModel subclasses always have __pydantic_generic_metadata__, so the `if not meta` early-return was dead code. Collapse it into a ternary and add a test for a plain (non-parameterized) ServerSentEvent subclass to reach the `not args` branch, bringing fastapi/sse.py to 100% coverage. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
The test only fetched /openapi.json, leaving the generator body unreachable. Add a /stream request inside the same TestClient context so the yield executes, bringing tests/test_sse.py to 100% coverage. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
|
There is a related discussion: #15102 That discussion goes one step further and wants to have a way to also constrain the
instead of just being able to say that the stream will contain |
|
@graipher nice -- I want that too. I have further modifications that make that kind of event-based data type discrimination possible, but I wanted to start with this narrower, smaller step first, to gauge appetite for these enhancements and understand any constraints on the implementation. That said -- I can put up the next steps that achieve that as a stacked draft PR so we can discuss the next steps also. |
Summary
ServerSentEventgeneric —ServerSentEvent[Data]now validates thatdatais an instance ofDataand emits acontentSchemafor that type in the OpenAPI specServerSentEvent(no type parameter) continues to accept any value includingNone, preserving full backward compatibilityget_sse_data_type()helper used by the routing layer to detect parameterizedServerSentEvent[Data]annotations and extractDatafor schema generationdocs_src/server_sent_events/tutorial006_py310.pyexample and corresponding testServerSentEvent" to the SSE tutorialMotivation
Before this change, yielding
ServerSentEventobjects from an SSE endpoint meant no data validation and no type information in the generated OpenAPI schema. Callers had to rely on documentation or convention to know what shape thedatafield would carry.With
ServerSentEvent[Item]:datareally is anItem(or raisesValidationErrorat construction time)contentSchemareferencingItem, so clients and SDK generators know the exact payload shapeHow it works
Pydantic's generic
BaseModelcreates a concrete subclass (not a_GenericAlias) when parameterized, soget_origen()returnsNone. The routing layer callsget_sse_data_type()which inspects__pydantic_generic_metadata__["args"]to extractData. The routing logic then branches:ServerSentEvent[Data]→ useDataasstream_item_type(feedscontentSchema)ServerSentEvent→ skip (nocontentSchema, same as before)model_config = ConfigDict(validate_default=True)ensures that omittingdataon a parameterized event raises aValidationErrorrather than silently storingNone.Example output
Checklist
tutorial006_py310.py) and tutorial test addedServerSentEventbehaves identically to beforeruff checkandruff formatpass with no changes