Sebastián Ramírez
3 months ago
committed by
GitHub
GitHub
23 changed files with 1681 additions and 37 deletions
@ -0,0 +1,120 @@ |
|||
# Server-Sent Events (SSE) { #server-sent-events-sse } |
|||
|
|||
You can stream data to the client using **Server-Sent Events** (SSE). |
|||
|
|||
This is similar to [Stream JSON Lines](stream-json-lines.md){.internal-link target=_blank}, but uses the `text/event-stream` format, which is supported natively by browsers with the <a href="https://developer.mozilla.org/en-US/docs/Web/API/EventSource" class="external-link" target="_blank">`EventSource` API</a>. |
|||
|
|||
/// info |
|||
|
|||
Added in FastAPI 0.135.0. |
|||
|
|||
/// |
|||
|
|||
## What are Server-Sent Events? { #what-are-server-sent-events } |
|||
|
|||
SSE is a standard for streaming data from the server to the client over HTTP. |
|||
|
|||
Each event is a small text block with "fields" like `data`, `event`, `id`, and `retry`, separated by blank lines. |
|||
|
|||
It looks like this: |
|||
|
|||
``` |
|||
data: {"name": "Portal Gun", "price": 999.99} |
|||
|
|||
data: {"name": "Plumbus", "price": 32.99} |
|||
|
|||
``` |
|||
|
|||
SSE is commonly used for AI chat streaming, live notifications, logs and observability, and other cases where the server pushes updates to the client. |
|||
|
|||
/// tip |
|||
|
|||
If you want to stream binary data, for example video or audio, check the advanced guide: [Stream Data](../advanced/stream-data.md){.internal-link target=_blank}. |
|||
|
|||
/// |
|||
|
|||
## Stream SSE with FastAPI { #stream-sse-with-fastapi } |
|||
|
|||
To stream SSE with FastAPI, use `yield` in your *path operation function* and set `response_class=EventSourceResponse`. |
|||
|
|||
Import `EventSourceResponse` from `fastapi.sse`: |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial001_py310.py ln[1:25] hl[4,22] *} |
|||
|
|||
Each yielded item is encoded as JSON and sent in the `data:` field of an SSE event. |
|||
|
|||
If you declare the return type as `AsyncIterable[Item]`, FastAPI will use it to **validate**, **document**, and **serialize** the data using Pydantic. |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial001_py310.py ln[1:25] hl[10:12,23] *} |
|||
|
|||
/// tip |
|||
|
|||
As Pydantic will serialize it in the **Rust** side, you will get much higher **performance** than if you don't declare a return type. |
|||
|
|||
/// |
|||
|
|||
### Non-async *path operation functions* { #non-async-path-operation-functions } |
|||
|
|||
You can also use regular `def` functions (without `async`), and use `yield` the same way. |
|||
|
|||
FastAPI will make sure it's run correctly so that it doesn't block the event loop. |
|||
|
|||
As in this case the function is not async, the right return type would be `Iterable[Item]`: |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial001_py310.py ln[28:31] hl[29] *} |
|||
|
|||
### No Return Type { #no-return-type } |
|||
|
|||
You can also omit the return type. FastAPI will use the [`jsonable_encoder`](./encoder.md){.internal-link target=_blank} to convert the data and send it. |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial001_py310.py ln[34:37] hl[35] *} |
|||
|
|||
## `ServerSentEvent` { #serversentevent } |
|||
|
|||
If you need to set SSE fields like `event`, `id`, `retry`, or `comment`, you can yield `ServerSentEvent` objects instead of plain data. |
|||
|
|||
Import `ServerSentEvent` from `fastapi.sse`: |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial002_py310.py hl[4,26] *} |
|||
|
|||
The `data` field is always encoded as JSON. You can pass any value that can be serialized as JSON, including Pydantic models. |
|||
|
|||
## Raw Data { #raw-data } |
|||
|
|||
If you need to send data **without** JSON encoding, use `raw_data` instead of `data`. |
|||
|
|||
This is useful for sending pre-formatted text, log lines, or special <dfn title="A value used to indicate a special condition or state">"sentinel"</dfn> values like `[DONE]`. |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial003_py310.py hl[17] *} |
|||
|
|||
/// note |
|||
|
|||
`data` and `raw_data` are mutually exclusive. You can only set one of them on each `ServerSentEvent`. |
|||
|
|||
/// |
|||
|
|||
## Resuming with `Last-Event-ID` { #resuming-with-last-event-id } |
|||
|
|||
When a browser reconnects after a connection drop, it sends the last received `id` in the `Last-Event-ID` header. |
|||
|
|||
You can read it as a header parameter and use it to resume the stream from where the client left off: |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial004_py310.py hl[25,27,31] *} |
|||
|
|||
## SSE with POST { #sse-with-post } |
|||
|
|||
SSE works with **any HTTP method**, not just `GET`. |
|||
|
|||
This is useful for protocols like <a href="https://modelcontextprotocol.io" class="external-link" target="_blank">MCP</a> that stream SSE over `POST`: |
|||
|
|||
{* ../../docs_src/server_sent_events/tutorial005_py310.py hl[14] *} |
|||
|
|||
## Technical Details { #technical-details } |
|||
|
|||
FastAPI implements some SSE best practices out of the box. |
|||
|
|||
* Send a **"keep alive" `ping` comment** every 15 seconds when there hasn't been any message, to prevent some proxies from closing the connection, as suggested in the <a href="https://html.spec.whatwg.org/multipage/server-sent-events.html#authoring-notes" class="external-link" target="_blank">HTML specification: Server-Sent Events</a>. |
|||
* Set the `Cache-Control: no-cache` header to **prevent caching** of the stream. |
|||
* Set a special header `X-Accel-Buffering: no` to **prevent buffering** in some proxies like Nginx. |
|||
|
|||
You don't have to do anything about it, it works out of the box. 🤓 |
|||
@ -0,0 +1,43 @@ |
|||
from collections.abc import AsyncIterable, Iterable |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.sse import EventSourceResponse |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None |
|||
|
|||
|
|||
items = [ |
|||
Item(name="Plumbus", description="A multi-purpose household device."), |
|||
Item(name="Portal Gun", description="A portal opening device."), |
|||
Item(name="Meeseeks Box", description="A box that summons a Meeseeks."), |
|||
] |
|||
|
|||
|
|||
@app.get("/items/stream", response_class=EventSourceResponse) |
|||
async def sse_items() -> AsyncIterable[Item]: |
|||
for item in items: |
|||
yield item |
|||
|
|||
|
|||
@app.get("/items/stream-no-async", response_class=EventSourceResponse) |
|||
def sse_items_no_async() -> Iterable[Item]: |
|||
for item in items: |
|||
yield item |
|||
|
|||
|
|||
@app.get("/items/stream-no-annotation", response_class=EventSourceResponse) |
|||
async def sse_items_no_annotation(): |
|||
for item in items: |
|||
yield item |
|||
|
|||
|
|||
@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse) |
|||
def sse_items_no_async_no_annotation(): |
|||
for item in items: |
|||
yield item |
|||
@ -0,0 +1,26 @@ |
|||
from collections.abc import AsyncIterable |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.sse import EventSourceResponse, ServerSentEvent |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
price: float |
|||
|
|||
|
|||
items = [ |
|||
Item(name="Plumbus", price=32.99), |
|||
Item(name="Portal Gun", price=999.99), |
|||
Item(name="Meeseeks Box", price=49.99), |
|||
] |
|||
|
|||
|
|||
@app.get("/items/stream", response_class=EventSourceResponse) |
|||
async def stream_items() -> AsyncIterable[ServerSentEvent]: |
|||
yield ServerSentEvent(comment="stream of item updates") |
|||
for i, item in enumerate(items): |
|||
yield ServerSentEvent(data=item, event="item_update", id=str(i + 1), retry=5000) |
|||
@ -0,0 +1,17 @@ |
|||
from collections.abc import AsyncIterable |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.sse import EventSourceResponse, ServerSentEvent |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/logs/stream", response_class=EventSourceResponse) |
|||
async def stream_logs() -> AsyncIterable[ServerSentEvent]: |
|||
logs = [ |
|||
"2025-01-01 INFO Application started", |
|||
"2025-01-01 DEBUG Connected to database", |
|||
"2025-01-01 WARN High memory usage detected", |
|||
] |
|||
for log_line in logs: |
|||
yield ServerSentEvent(raw_data=log_line) |
|||
@ -0,0 +1,31 @@ |
|||
from collections.abc import AsyncIterable |
|||
from typing import Annotated |
|||
|
|||
from fastapi import FastAPI, Header |
|||
from fastapi.sse import EventSourceResponse, ServerSentEvent |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
price: float |
|||
|
|||
|
|||
items = [ |
|||
Item(name="Plumbus", price=32.99), |
|||
Item(name="Portal Gun", price=999.99), |
|||
Item(name="Meeseeks Box", price=49.99), |
|||
] |
|||
|
|||
|
|||
@app.get("/items/stream", response_class=EventSourceResponse) |
|||
async def stream_items( |
|||
last_event_id: Annotated[int | None, Header()] = None, |
|||
) -> AsyncIterable[ServerSentEvent]: |
|||
start = last_event_id + 1 if last_event_id is not None else 0 |
|||
for i, item in enumerate(items): |
|||
if i < start: |
|||
continue |
|||
yield ServerSentEvent(data=item, id=str(i)) |
|||
@ -0,0 +1,19 @@ |
|||
from collections.abc import AsyncIterable |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.sse import EventSourceResponse, ServerSentEvent |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Prompt(BaseModel): |
|||
text: str |
|||
|
|||
|
|||
@app.post("/chat/stream", response_class=EventSourceResponse) |
|||
async def stream_chat(prompt: Prompt) -> AsyncIterable[ServerSentEvent]: |
|||
words = prompt.text.split() |
|||
for word in words: |
|||
yield ServerSentEvent(data=word, event="token") |
|||
yield ServerSentEvent(raw_data="[DONE]", event="done") |
|||
@ -0,0 +1,222 @@ |
|||
from typing import Annotated, Any |
|||
|
|||
from annotated_doc import Doc |
|||
from pydantic import AfterValidator, BaseModel, Field, model_validator |
|||
from starlette.responses import StreamingResponse |
|||
|
|||
# Canonical SSE event schema matching the OpenAPI 3.2 spec |
|||
# (Section 4.14.4 "Special Considerations for Server-Sent Events") |
|||
_SSE_EVENT_SCHEMA: dict[str, Any] = { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": {"type": "integer", "minimum": 0}, |
|||
}, |
|||
} |
|||
|
|||
|
|||
class EventSourceResponse(StreamingResponse): |
|||
"""Streaming response with `text/event-stream` media type. |
|||
|
|||
Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield` |
|||
to enable Server Sent Events (SSE) responses. |
|||
|
|||
Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible |
|||
with protocols like MCP that stream SSE over `POST`. |
|||
|
|||
The actual encoding logic lives in the FastAPI routing layer. This class |
|||
serves mainly as a marker and sets the correct `Content-Type`. |
|||
""" |
|||
|
|||
media_type = "text/event-stream" |
|||
|
|||
|
|||
def _check_id_no_null(v: str | None) -> str | None: |
|||
if v is not None and "\0" in v: |
|||
raise ValueError("SSE 'id' must not contain null characters") |
|||
return v |
|||
|
|||
|
|||
class ServerSentEvent(BaseModel): |
|||
"""Represents a single Server-Sent Event. |
|||
|
|||
When `yield`ed from a *path operation function* that uses |
|||
`response_class=EventSourceResponse`, each `ServerSentEvent` is encoded |
|||
into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream) |
|||
(`text/event-stream`). |
|||
|
|||
If you yield a plain object (dict, Pydantic model, etc.) instead, it is |
|||
automatically JSON-encoded and sent as the `data:` field. |
|||
|
|||
All `data` values **including plain strings** are JSON-serialized. |
|||
|
|||
For example, `data="hello"` produces `data: "hello"` on the wire (with |
|||
quotes). |
|||
""" |
|||
|
|||
data: Annotated[ |
|||
Any, |
|||
Doc( |
|||
""" |
|||
The event payload. |
|||
|
|||
Can be any JSON-serializable value: a Pydantic model, dict, list, |
|||
string, number, etc. It is **always** serialized to JSON: strings |
|||
are quoted (`"hello"` becomes `data: "hello"` on the wire). |
|||
|
|||
Mutually exclusive with `raw_data`. |
|||
""" |
|||
), |
|||
] = None |
|||
raw_data: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Raw string to send as the `data:` field **without** JSON encoding. |
|||
|
|||
Use this when you need to send pre-formatted text, HTML fragments, |
|||
CSV lines, or any non-JSON payload. The string is placed directly |
|||
into the `data:` field as-is. |
|||
|
|||
Mutually exclusive with `data`. |
|||
""" |
|||
), |
|||
] = None |
|||
event: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Optional event type name. |
|||
|
|||
Maps to `addEventListener(event, ...)` on the browser. When omitted, |
|||
the browser dispatches on the generic `message` event. |
|||
""" |
|||
), |
|||
] = None |
|||
id: Annotated[ |
|||
str | None, |
|||
AfterValidator(_check_id_no_null), |
|||
Doc( |
|||
""" |
|||
Optional event ID. |
|||
|
|||
The browser sends this value back as the `Last-Event-ID` header on |
|||
automatic reconnection. **Must not contain null (`\\0`) characters.** |
|||
""" |
|||
), |
|||
] = None |
|||
retry: Annotated[ |
|||
int | None, |
|||
Field(ge=0), |
|||
Doc( |
|||
""" |
|||
Optional reconnection time in **milliseconds**. |
|||
|
|||
Tells the browser how long to wait before reconnecting after the |
|||
connection is lost. Must be a non-negative integer. |
|||
""" |
|||
), |
|||
] = None |
|||
comment: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Optional comment line(s). |
|||
|
|||
Comment lines start with `:` in the SSE wire format and are ignored by |
|||
`EventSource` clients. Useful for keep-alive pings to prevent |
|||
proxy/load-balancer timeouts. |
|||
""" |
|||
), |
|||
] = None |
|||
|
|||
@model_validator(mode="after") |
|||
def _check_data_exclusive(self) -> "ServerSentEvent": |
|||
if self.data is not None and self.raw_data is not None: |
|||
raise ValueError( |
|||
"Cannot set both 'data' and 'raw_data' on the same " |
|||
"ServerSentEvent. Use 'data' for JSON-serialized payloads " |
|||
"or 'raw_data' for pre-formatted strings." |
|||
) |
|||
return self |
|||
|
|||
|
|||
def format_sse_event( |
|||
*, |
|||
data_str: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Pre-serialized data string to use as the `data:` field. |
|||
""" |
|||
), |
|||
] = None, |
|||
event: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Optional event type name (`event:` field). |
|||
""" |
|||
), |
|||
] = None, |
|||
id: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Optional event ID (`id:` field). |
|||
""" |
|||
), |
|||
] = None, |
|||
retry: Annotated[ |
|||
int | None, |
|||
Doc( |
|||
""" |
|||
Optional reconnection time in milliseconds (`retry:` field). |
|||
""" |
|||
), |
|||
] = None, |
|||
comment: Annotated[ |
|||
str | None, |
|||
Doc( |
|||
""" |
|||
Optional comment line(s) (`:` prefix). |
|||
""" |
|||
), |
|||
] = None, |
|||
) -> bytes: |
|||
"""Build SSE wire-format bytes from **pre-serialized** data. |
|||
|
|||
The result always ends with `\n\n` (the event terminator). |
|||
""" |
|||
lines: list[str] = [] |
|||
|
|||
if comment is not None: |
|||
for line in comment.splitlines(): |
|||
lines.append(f": {line}") |
|||
|
|||
if event is not None: |
|||
lines.append(f"event: {event}") |
|||
|
|||
if data_str is not None: |
|||
for line in data_str.splitlines(): |
|||
lines.append(f"data: {line}") |
|||
|
|||
if id is not None: |
|||
lines.append(f"id: {id}") |
|||
|
|||
if retry is not None: |
|||
lines.append(f"retry: {retry}") |
|||
|
|||
lines.append("") |
|||
lines.append("") |
|||
return "\n".join(lines).encode("utf-8") |
|||
|
|||
|
|||
# Keep-alive comment, per the SSE spec recommendation |
|||
KEEPALIVE_COMMENT = b": ping\n\n" |
|||
|
|||
# Seconds between keep-alive pings when a generator is idle. |
|||
# Private but importable so tests can monkeypatch it. |
|||
_PING_INTERVAL: float = 15.0 |
|||
@ -0,0 +1,318 @@ |
|||
import asyncio |
|||
import time |
|||
from collections.abc import AsyncIterable, Iterable |
|||
|
|||
import fastapi.routing |
|||
import pytest |
|||
from fastapi import APIRouter, FastAPI |
|||
from fastapi.responses import EventSourceResponse |
|||
from fastapi.sse import ServerSentEvent |
|||
from fastapi.testclient import TestClient |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
|
|||
|
|||
items = [ |
|||
Item(name="Plumbus", description="A multi-purpose household device."), |
|||
Item(name="Portal Gun", description="A portal opening device."), |
|||
Item(name="Meeseeks Box", description="A box that summons a Meeseeks."), |
|||
] |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/stream", response_class=EventSourceResponse) |
|||
async def sse_items() -> AsyncIterable[Item]: |
|||
for item in items: |
|||
yield item |
|||
|
|||
|
|||
@app.get("/items/stream-sync", response_class=EventSourceResponse) |
|||
def sse_items_sync() -> Iterable[Item]: |
|||
yield from items |
|||
|
|||
|
|||
@app.get("/items/stream-no-annotation", response_class=EventSourceResponse) |
|||
async def sse_items_no_annotation(): |
|||
for item in items: |
|||
yield item |
|||
|
|||
|
|||
@app.get("/items/stream-sync-no-annotation", response_class=EventSourceResponse) |
|||
def sse_items_sync_no_annotation(): |
|||
yield from items |
|||
|
|||
|
|||
@app.get("/items/stream-dict", response_class=EventSourceResponse) |
|||
async def sse_items_dict(): |
|||
for item in items: |
|||
yield {"name": item.name, "description": item.description} |
|||
|
|||
|
|||
@app.get("/items/stream-sse-event", response_class=EventSourceResponse) |
|||
async def sse_items_event(): |
|||
yield ServerSentEvent(data="hello", event="greeting", id="1") |
|||
yield ServerSentEvent(data={"key": "value"}, event="json-data", id="2") |
|||
yield ServerSentEvent(comment="just a comment") |
|||
yield ServerSentEvent(data="retry-test", retry=5000) |
|||
|
|||
|
|||
@app.get("/items/stream-mixed", response_class=EventSourceResponse) |
|||
async def sse_items_mixed() -> AsyncIterable[Item]: |
|||
yield items[0] |
|||
yield ServerSentEvent(data="custom-event", event="special") |
|||
yield items[1] |
|||
|
|||
|
|||
@app.get("/items/stream-string", response_class=EventSourceResponse) |
|||
async def sse_items_string(): |
|||
yield ServerSentEvent(data="plain text data") |
|||
|
|||
|
|||
@app.post("/items/stream-post", response_class=EventSourceResponse) |
|||
async def sse_items_post() -> AsyncIterable[Item]: |
|||
for item in items: |
|||
yield item |
|||
|
|||
|
|||
@app.get("/items/stream-raw", response_class=EventSourceResponse) |
|||
async def sse_items_raw(): |
|||
yield ServerSentEvent(raw_data="plain text without quotes") |
|||
yield ServerSentEvent(raw_data="<div>html fragment</div>", event="html") |
|||
yield ServerSentEvent(raw_data="cpu,87.3,1709145600", event="csv") |
|||
|
|||
|
|||
router = APIRouter() |
|||
|
|||
|
|||
@router.get("/events", response_class=EventSourceResponse) |
|||
async def stream_events(): |
|||
yield {"msg": "hello"} |
|||
yield {"msg": "world"} |
|||
|
|||
|
|||
app.include_router(router, prefix="/api") |
|||
|
|||
|
|||
@pytest.fixture(name="client") |
|||
def client_fixture(): |
|||
with TestClient(app) as c: |
|||
yield c |
|||
|
|||
|
|||
def test_async_generator_with_model(client: TestClient): |
|||
response = client.get("/items/stream") |
|||
assert response.status_code == 200 |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
assert response.headers["cache-control"] == "no-cache" |
|||
assert response.headers["x-accel-buffering"] == "no" |
|||
|
|||
lines = response.text.strip().split("\n") |
|||
data_lines = [line for line in lines if line.startswith("data: ")] |
|||
assert len(data_lines) == 3 |
|||
assert '"name":"Plumbus"' in data_lines[0] or '"name": "Plumbus"' in data_lines[0] |
|||
assert ( |
|||
'"name":"Portal Gun"' in data_lines[1] |
|||
or '"name": "Portal Gun"' in data_lines[1] |
|||
) |
|||
assert ( |
|||
'"name":"Meeseeks Box"' in data_lines[2] |
|||
or '"name": "Meeseeks Box"' in data_lines[2] |
|||
) |
|||
|
|||
|
|||
def test_sync_generator_with_model(client: TestClient): |
|||
response = client.get("/items/stream-sync") |
|||
assert response.status_code == 200 |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
|
|||
def test_async_generator_no_annotation(client: TestClient): |
|||
response = client.get("/items/stream-no-annotation") |
|||
assert response.status_code == 200 |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
|
|||
def test_sync_generator_no_annotation(client: TestClient): |
|||
response = client.get("/items/stream-sync-no-annotation") |
|||
assert response.status_code == 200 |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
|
|||
def test_dict_items(client: TestClient): |
|||
response = client.get("/items/stream-dict") |
|||
assert response.status_code == 200 |
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
assert '"name"' in data_lines[0] |
|||
|
|||
|
|||
def test_post_method_sse(client: TestClient): |
|||
"""SSE should work with POST (needed for MCP compatibility).""" |
|||
response = client.post("/items/stream-post") |
|||
assert response.status_code == 200 |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
|
|||
def test_sse_events_with_fields(client: TestClient): |
|||
response = client.get("/items/stream-sse-event") |
|||
assert response.status_code == 200 |
|||
text = response.text |
|||
|
|||
assert "event: greeting\n" in text |
|||
assert 'data: "hello"\n' in text |
|||
assert "id: 1\n" in text |
|||
|
|||
assert "event: json-data\n" in text |
|||
assert "id: 2\n" in text |
|||
assert 'data: {"key": "value"}\n' in text |
|||
|
|||
assert ": just a comment\n" in text |
|||
|
|||
assert "retry: 5000\n" in text |
|||
assert 'data: "retry-test"\n' in text |
|||
|
|||
|
|||
def test_mixed_plain_and_sse_events(client: TestClient): |
|||
response = client.get("/items/stream-mixed") |
|||
assert response.status_code == 200 |
|||
text = response.text |
|||
|
|||
assert "event: special\n" in text |
|||
assert 'data: "custom-event"\n' in text |
|||
assert '"name"' in text |
|||
|
|||
|
|||
def test_string_data_json_encoded(client: TestClient): |
|||
"""Strings are always JSON-encoded (quoted).""" |
|||
response = client.get("/items/stream-string") |
|||
assert response.status_code == 200 |
|||
assert 'data: "plain text data"\n' in response.text |
|||
|
|||
|
|||
def test_server_sent_event_null_id_rejected(): |
|||
with pytest.raises(ValueError, match="null"): |
|||
ServerSentEvent(data="test", id="has\0null") |
|||
|
|||
|
|||
def test_server_sent_event_negative_retry_rejected(): |
|||
with pytest.raises(ValueError): |
|||
ServerSentEvent(data="test", retry=-1) |
|||
|
|||
|
|||
def test_server_sent_event_float_retry_rejected(): |
|||
with pytest.raises(ValueError): |
|||
ServerSentEvent(data="test", retry=1.5) # type: ignore[arg-type] |
|||
|
|||
|
|||
def test_raw_data_sent_without_json_encoding(client: TestClient): |
|||
"""raw_data is sent as-is, not JSON-encoded.""" |
|||
response = client.get("/items/stream-raw") |
|||
assert response.status_code == 200 |
|||
text = response.text |
|||
|
|||
# raw_data should appear without JSON quotes |
|||
assert "data: plain text without quotes\n" in text |
|||
# Not JSON-quoted |
|||
assert 'data: "plain text without quotes"' not in text |
|||
|
|||
assert "event: html\n" in text |
|||
assert "data: <div>html fragment</div>\n" in text |
|||
|
|||
assert "event: csv\n" in text |
|||
assert "data: cpu,87.3,1709145600\n" in text |
|||
|
|||
|
|||
def test_data_and_raw_data_mutually_exclusive(): |
|||
"""Cannot set both data and raw_data.""" |
|||
with pytest.raises(ValueError, match="Cannot set both"): |
|||
ServerSentEvent(data="json", raw_data="raw") |
|||
|
|||
|
|||
def test_sse_on_router_included_in_app(client: TestClient): |
|||
response = client.get("/api/events") |
|||
assert response.status_code == 200 |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 2 |
|||
|
|||
|
|||
# Keepalive ping tests |
|||
|
|||
|
|||
keepalive_app = FastAPI() |
|||
|
|||
|
|||
@keepalive_app.get("/slow-async", response_class=EventSourceResponse) |
|||
async def slow_async_stream(): |
|||
yield {"n": 1} |
|||
# Sleep longer than the (monkeypatched) ping interval so a keepalive |
|||
# comment is emitted before the next item. |
|||
await asyncio.sleep(0.3) |
|||
yield {"n": 2} |
|||
|
|||
|
|||
@keepalive_app.get("/slow-sync", response_class=EventSourceResponse) |
|||
def slow_sync_stream(): |
|||
yield {"n": 1} |
|||
time.sleep(0.3) |
|||
yield {"n": 2} |
|||
|
|||
|
|||
def test_keepalive_ping_async(monkeypatch: pytest.MonkeyPatch): |
|||
monkeypatch.setattr(fastapi.routing, "_PING_INTERVAL", 0.05) |
|||
with TestClient(keepalive_app) as c: |
|||
response = c.get("/slow-async") |
|||
assert response.status_code == 200 |
|||
text = response.text |
|||
# The keepalive comment ": ping" should appear between the two data events |
|||
assert ": ping\n" in text |
|||
data_lines = [line for line in text.split("\n") if line.startswith("data: ")] |
|||
assert len(data_lines) == 2 |
|||
|
|||
|
|||
def test_keepalive_ping_sync(monkeypatch: pytest.MonkeyPatch): |
|||
monkeypatch.setattr(fastapi.routing, "_PING_INTERVAL", 0.05) |
|||
with TestClient(keepalive_app) as c: |
|||
response = c.get("/slow-sync") |
|||
assert response.status_code == 200 |
|||
text = response.text |
|||
assert ": ping\n" in text |
|||
data_lines = [line for line in text.split("\n") if line.startswith("data: ")] |
|||
assert len(data_lines) == 2 |
|||
|
|||
|
|||
def test_no_keepalive_when_fast(client: TestClient): |
|||
"""No keepalive comment when items arrive quickly.""" |
|||
response = client.get("/items/stream") |
|||
assert response.status_code == 200 |
|||
# KEEPALIVE_COMMENT is ": ping\n\n". |
|||
assert ": ping\n" not in response.text |
|||
@ -0,0 +1,191 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
pytest.param("tutorial001_py310"), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}") |
|||
|
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
@pytest.mark.parametrize( |
|||
"path", |
|||
[ |
|||
"/items/stream", |
|||
"/items/stream-no-async", |
|||
"/items/stream-no-annotation", |
|||
"/items/stream-no-async-no-annotation", |
|||
], |
|||
) |
|||
def test_stream_items(client: TestClient, path: str): |
|||
response = client.get(path) |
|||
assert response.status_code == 200, response.text |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
|
|||
def test_openapi_schema(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/items/stream": { |
|||
"get": { |
|||
"summary": "Sse Items", |
|||
"operationId": "sse_items_items_stream_get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": { |
|||
"type": "string", |
|||
"contentMediaType": "application/json", |
|||
"contentSchema": { |
|||
"$ref": "#/components/schemas/Item" |
|||
}, |
|||
}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
"required": ["data"], |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
"/items/stream-no-async": { |
|||
"get": { |
|||
"summary": "Sse Items No Async", |
|||
"operationId": "sse_items_no_async_items_stream_no_async_get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": { |
|||
"type": "string", |
|||
"contentMediaType": "application/json", |
|||
"contentSchema": { |
|||
"$ref": "#/components/schemas/Item" |
|||
}, |
|||
}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
"required": ["data"], |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
"/items/stream-no-annotation": { |
|||
"get": { |
|||
"summary": "Sse Items No Annotation", |
|||
"operationId": "sse_items_no_annotation_items_stream_no_annotation_get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
"/items/stream-no-async-no-annotation": { |
|||
"get": { |
|||
"summary": "Sse Items No Async No Annotation", |
|||
"operationId": "sse_items_no_async_no_annotation_items_stream_no_async_no_annotation_get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"Item": { |
|||
"properties": { |
|||
"name": {"type": "string", "title": "Name"}, |
|||
"description": { |
|||
"anyOf": [ |
|||
{"type": "string"}, |
|||
{"type": "null"}, |
|||
], |
|||
"title": "Description", |
|||
}, |
|||
}, |
|||
"type": "object", |
|||
"required": ["name", "description"], |
|||
"title": "Item", |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
@ -0,0 +1,83 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
pytest.param("tutorial002_py310"), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}") |
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
def test_stream_items(client: TestClient): |
|||
response = client.get("/items/stream") |
|||
assert response.status_code == 200, response.text |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
|
|||
lines = response.text.strip().split("\n") |
|||
|
|||
# First event is a comment-only event |
|||
assert lines[0] == ": stream of item updates" |
|||
|
|||
# Remaining lines contain event:, data:, id:, retry: fields |
|||
event_lines = [line for line in lines if line.startswith("event: ")] |
|||
assert len(event_lines) == 3 |
|||
assert all(line == "event: item_update" for line in event_lines) |
|||
|
|||
data_lines = [line for line in lines if line.startswith("data: ")] |
|||
assert len(data_lines) == 3 |
|||
|
|||
id_lines = [line for line in lines if line.startswith("id: ")] |
|||
assert id_lines == ["id: 1", "id: 2", "id: 3"] |
|||
|
|||
retry_lines = [line for line in lines if line.startswith("retry: ")] |
|||
assert len(retry_lines) == 3 |
|||
assert all(line == "retry: 5000" for line in retry_lines) |
|||
|
|||
|
|||
def test_openapi_schema(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/items/stream": { |
|||
"get": { |
|||
"summary": "Stream Items", |
|||
"operationId": "stream_items_items_stream_get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
@ -0,0 +1,73 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
pytest.param("tutorial003_py310"), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}") |
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
def test_stream_logs(client: TestClient): |
|||
response = client.get("/logs/stream") |
|||
assert response.status_code == 200, response.text |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
# raw_data is sent without JSON encoding (no quotes around the string) |
|||
assert data_lines[0] == "data: 2025-01-01 INFO Application started" |
|||
assert data_lines[1] == "data: 2025-01-01 DEBUG Connected to database" |
|||
assert data_lines[2] == "data: 2025-01-01 WARN High memory usage detected" |
|||
|
|||
|
|||
def test_openapi_schema(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/logs/stream": { |
|||
"get": { |
|||
"summary": "Stream Logs", |
|||
"operationId": "stream_logs_logs_stream_get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
@ -0,0 +1,164 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
pytest.param("tutorial004_py310"), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}") |
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
def test_stream_all_items(client: TestClient): |
|||
response = client.get("/items/stream") |
|||
assert response.status_code == 200, response.text |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 3 |
|||
|
|||
id_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("id: ") |
|||
] |
|||
assert id_lines == ["id: 0", "id: 1", "id: 2"] |
|||
|
|||
|
|||
def test_resume_from_last_event_id(client: TestClient): |
|||
response = client.get( |
|||
"/items/stream", |
|||
headers={"last-event-id": "0"}, |
|||
) |
|||
assert response.status_code == 200, response.text |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 2 |
|||
|
|||
id_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("id: ") |
|||
] |
|||
assert id_lines == ["id: 1", "id: 2"] |
|||
|
|||
|
|||
def test_resume_from_last_item(client: TestClient): |
|||
response = client.get( |
|||
"/items/stream", |
|||
headers={"last-event-id": "1"}, |
|||
) |
|||
assert response.status_code == 200, response.text |
|||
|
|||
data_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("data: ") |
|||
] |
|||
assert len(data_lines) == 1 |
|||
|
|||
id_lines = [ |
|||
line for line in response.text.strip().split("\n") if line.startswith("id: ") |
|||
] |
|||
assert id_lines == ["id: 2"] |
|||
|
|||
|
|||
def test_openapi_schema(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/items/stream": { |
|||
"get": { |
|||
"summary": "Stream Items", |
|||
"operationId": "stream_items_items_stream_get", |
|||
"parameters": [ |
|||
{ |
|||
"name": "last-event-id", |
|||
"in": "header", |
|||
"required": False, |
|||
"schema": { |
|||
"anyOf": [{"type": "integer"}, {"type": "null"}], |
|||
"title": "Last-Event-Id", |
|||
}, |
|||
} |
|||
], |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"HTTPValidationError": { |
|||
"properties": { |
|||
"detail": { |
|||
"items": { |
|||
"$ref": "#/components/schemas/ValidationError" |
|||
}, |
|||
"type": "array", |
|||
"title": "Detail", |
|||
} |
|||
}, |
|||
"type": "object", |
|||
"title": "HTTPValidationError", |
|||
}, |
|||
"ValidationError": { |
|||
"properties": { |
|||
"loc": { |
|||
"items": { |
|||
"anyOf": [{"type": "string"}, {"type": "integer"}] |
|||
}, |
|||
"type": "array", |
|||
"title": "Location", |
|||
}, |
|||
"msg": {"type": "string", "title": "Message"}, |
|||
"type": {"type": "string", "title": "Error Type"}, |
|||
"input": {"title": "Input"}, |
|||
"ctx": {"type": "object", "title": "Context"}, |
|||
}, |
|||
"type": "object", |
|||
"required": ["loc", "msg", "type"], |
|||
"title": "ValidationError", |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
@ -0,0 +1,141 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
pytest.param("tutorial005_py310"), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.server_sent_events.{request.param}") |
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
def test_stream_chat(client: TestClient): |
|||
response = client.post( |
|||
"/chat/stream", |
|||
json={"text": "hello world"}, |
|||
) |
|||
assert response.status_code == 200, response.text |
|||
assert response.headers["content-type"] == "text/event-stream; charset=utf-8" |
|||
|
|||
lines = response.text.strip().split("\n") |
|||
|
|||
event_lines = [line for line in lines if line.startswith("event: ")] |
|||
assert event_lines == [ |
|||
"event: token", |
|||
"event: token", |
|||
"event: done", |
|||
] |
|||
|
|||
data_lines = [line for line in lines if line.startswith("data: ")] |
|||
assert data_lines == [ |
|||
'data: "hello"', |
|||
'data: "world"', |
|||
"data: [DONE]", |
|||
] |
|||
|
|||
|
|||
def test_openapi_schema(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/chat/stream": { |
|||
"post": { |
|||
"summary": "Stream Chat", |
|||
"operationId": "stream_chat_chat_stream_post", |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/Prompt"} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"text/event-stream": { |
|||
"itemSchema": { |
|||
"type": "object", |
|||
"properties": { |
|||
"data": {"type": "string"}, |
|||
"event": {"type": "string"}, |
|||
"id": {"type": "string"}, |
|||
"retry": { |
|||
"type": "integer", |
|||
"minimum": 0, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"HTTPValidationError": { |
|||
"properties": { |
|||
"detail": { |
|||
"items": { |
|||
"$ref": "#/components/schemas/ValidationError" |
|||
}, |
|||
"type": "array", |
|||
"title": "Detail", |
|||
} |
|||
}, |
|||
"type": "object", |
|||
"title": "HTTPValidationError", |
|||
}, |
|||
"Prompt": { |
|||
"properties": {"text": {"type": "string", "title": "Text"}}, |
|||
"type": "object", |
|||
"required": ["text"], |
|||
"title": "Prompt", |
|||
}, |
|||
"ValidationError": { |
|||
"properties": { |
|||
"loc": { |
|||
"items": { |
|||
"anyOf": [{"type": "string"}, {"type": "integer"}] |
|||
}, |
|||
"type": "array", |
|||
"title": "Location", |
|||
}, |
|||
"msg": {"type": "string", "title": "Message"}, |
|||
"type": {"type": "string", "title": "Error Type"}, |
|||
"input": {"title": "Input"}, |
|||
"ctx": {"type": "object", "title": "Context"}, |
|||
}, |
|||
"type": "object", |
|||
"required": ["loc", "msg", "type"], |
|||
"title": "ValidationError", |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
Loading…
Reference in new issue