Потоковые ответы и WebSockets

Урок разбирает три способа не отдавать ответ «целиком и сразу»: StreamingResponse, Server-Sent Events и WebSocket — и помогает выбрать нужный.

Потоковая отдача — отправка ответа клиенту частями по мере готовности, вместо того чтобы собрать всё в памяти и отдать одним куском; для живой двусторонней связи существует отдельный протокол — WebSocket.

Обычный обработчик FastAPI формирует ответ полностью и затем отдаёт его. Это плохо работает в трёх случаях: ответ огромный (не влезает в память комфортно), данные появляются постепенно (отчёт, токены LLM), или нужна живая двусторонняя связь (чат, игра). Для каждого случая есть свой инструмент.

Разобраться важно, потому что выбрать не тот механизм — значит либо съесть память на больших ответах, либо городить костыли поверх HTTP там, где нужен WebSocket, либо наоборот тащить WebSocket туда, где хватило бы простого потока.

Зачем это на практике

Три типичные задачи. Первая: отдать CSV на миллион строк — собирать его в памяти расточительно, лучше слать построчно. Вторая: показывать прогресс генерации или ответ модели по мере появления слов — клиенту нужны обновления «на лету», но только в одну сторону, от сервера. Третья: чат или совместный редактор — здесь сообщения летят в обе стороны постоянно. Это ровно три разных инструмента: StreamingResponse, SSE и WebSocket.

StreamingResponse: ответ частями

StreamingResponse принимает генератор и отправляет то, что он выдаёт (yield), кусками, не держа всё в памяти. Подходит для больших файлов и любой постепенной генерации. Пример импортирует FastAPI — он для чтения.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

def generate_csv():
    yield "id,name\n"
    for i in range(1, 1_000_000):
        yield f"{i},user{i}\n"      # каждая строка уходит сразу, память не растёт

@app.get("/export")
def export():
    return StreamingResponse(generate_csv(), media_type="text/csv")

Идея генератора — на чистом Python (исполнимо в браузере): функция отдаёт куски по одному, а не собирает список целиком.

def generate_csv():
    yield "id,name\n"
    for i in range(1, 4):
        yield f"{i},user{i}\n"

for chunk in generate_csv():
    print(repr(chunk))

Вывод:

'id,name\n'
'1,user1\n'
'2,user2\n'
'3,user3\n'

Каждый yield — отдельный кусок, который во FastAPI ушёл бы клиенту немедленно. В реальном экспорте таких кусков миллион, но в памяти одновременно — лишь один. Это и есть смысл потоковой отдачи: постоянное потребление памяти независимо от размера ответа.

Server-Sent Events: односторонние обновления

Server-Sent Events (SSE) — это StreamingResponse особого формата: сервер держит соединение открытым и шлёт клиенту события по мере появления. Связь односторонняя (только сервер → клиент), работает поверх обычного HTTP, а браузер умеет принимать такие события через EventSource и сам переподключается при обрыве. Формат строгий: каждое сообщение — строка data: ..., завершённая двумя переводами строки.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream():
    for i in range(1, 4):
        yield f"data: сообщение {i}\n\n"   # формат SSE: data: ... + пустая строка
        await asyncio.sleep(1)

@app.get("/events")
async def events():
    return StreamingResponse(event_stream(), media_type="text/event-stream")

Сам формат кадров SSE можно собрать обычным Python и убедиться, что каждое событие оканчивается пустой строкой (\n\n):

def sse_events():
    for i in range(1, 4):
        yield f"data: сообщение {i}\n\n"

for chunk in sse_events():
    print(repr(chunk))

Вывод:

'data: сообщение 1\n\n'
'data: сообщение 2\n\n'
'data: сообщение 3\n\n'

SSE идеален для прогресс-баров, лент обновлений, дашбордов, потоковой выдачи токенов LLM — всего, где данные идут только от сервера. Он проще WebSocket: обычный HTTP, авто-переподключение из коробки, не нужен особый протокол.

WebSocket: двунаправленная связь

Когда сообщения должны идти в обе стороны и постоянно — чат, многопользовательская игра, совместное редактирование, торговый терминал — нужен WebSocket. Это отдельный протокол: клиент и сервер один раз «жмут руки» поверх HTTP, после чего держат постоянное двустороннее соединение и шлют сообщения в любую сторону без накладных расходов на новые запросы. FastAPI поддерживает его через декоратор @app.websocket. Пример для чтения:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()                      # принимаем соединение
    try:
        while True:
            msg = await ws.receive_text()  # ждём сообщение от клиента
            await ws.send_text(f"эхо: {msg}")  # отвечаем в ту же связь
    except WebSocketDisconnect:
        pass                               # клиент отключился

Обратите внимание на цикл while True с await receive_text(): соединение живёт долго, и обе стороны обмениваются сообщениями, пока кто-то не отключится. Это невозможно в обычном HTTP-запросе, который живёт один цикл «запрос-ответ».

Что выбрать

Правило выбора: нужна двусторонность — WebSocket; данные только от сервера и хватает HTTP — SSE; это просто большой/постепенный ответ на один запрос — StreamingResponse.

Как это работает под капотом

StreamingResponse и SSE — это обычный HTTP-ответ с заголовком, говорящим «тело придёт частями» (chunked transfer / открытое соединение). Сервер не закрывает соединение и пишет в него куски по мере yield; клиент читает их по мере поступления. Никакого нового протокола — это всё ещё один HTTP-ответ, просто растянутый во времени.

WebSocket начинается с HTTP-запроса с заголовком Upgrade: websocket. Сервер отвечает 101 Switching Protocols, и дальше тот же TCP-сокет используется уже не для HTTP, а для двустороннего обмена кадрами WebSocket. Поэтому WebSocket — полноценный отдельный протокол поверх того же соединения, а не «HTTP с потоком». Поскольку и SSE, и WebSocket держат соединение открытым подолгу, помните прошлые уроки: внутри их обработчиков нельзя блокировать event loop синхронным кодом.

Частые ошибки

Собирать огромный ответ в память вместо StreamingResponse. Сформировать список на миллион строк, склеить в одну строку и вернуть — пик памяти на каждого клиента. Потоковая отдача держит память постоянной.

Брать WebSocket там, где хватает SSE. Если данные идут только от сервера (прогресс, лента), WebSocket — лишняя сложность: нет авто-переподключения из коробки, тяжелее проксировать и масштабировать. SSE проще и достаточно.

Нарушать формат SSE. Каждое событие обязано заканчиваться пустой строкой (\n\n), а тело идти как data: .... Забудешь двойной перевод строки — браузерный EventSource не разберёт поток.

Блокировать цикл в долгоживущем стриме. SSE и WebSocket держат соединение надолго; синхронный блокирующий вызов в их обработчике остановит весь event loop и заденет другие соединения. Используйте асинхронные вызовы или уводите блокировку в поток.

Забыть обработать отключение WebSocket. Клиент может уйти в любой момент; без перехвата WebSocketDisconnect вы получите исключение и, возможно, утечку ресурсов (незакрытые подписки, записи о клиенте).

Итоги

• StreamingResponse отдаёт один ответ частями по мере yield — для больших файлов и постепенной генерации, память остаётся постоянной.
• SSE — это StreamingResponse в формате text/event-stream: односторонний поток событий сервер → клиент поверх обычного HTTP, с авто-переподключением.
• WebSocket — отдельный протокол для двусторонней живой связи (чат, игры, совместная работа); начинается с HTTP-апгрейда, дальше держит постоянный сокет.
• Выбор: двусторонность → WebSocket; обновления только от сервера → SSE; большой/постепенный ответ на один запрос → StreamingResponse.
• SSE и WebSocket живут долго — не блокируйте в них event loop и обрабатывайте отключение клиента (WebSocketDisconnect).