SSE в FastAPI: как продолжить поток после обрыва по Last-Event-ID

Страница показывает, как готовится выгрузка. Браузер увидел «проверяем данные», затем ноутбук переключился на другую сеть. К моменту переподключения сервер уже собрал файл. Если маршрут умеет только подписывать клиента на новые события, браузер будет ждать следующего сообщения и может так и не показать кнопку скачивания.

При отправке каждого обновления SSE позволяет приложить id. При автоматическом переподключении EventSource присылает последний полученный идентификатор в Last-Event-ID. Браузер сообщает позицию в потоке, а сервер решает, какие изменения отправить дальше. Для этого идентификатор должен относиться к записи в журнале конкретной задачи. Счётчик внутри обработчика запроса при каждом подключении начнёт отсчёт заново.

Дайте каждому клиенту последовательность, которую можно повторить

В этом примере у задачи четыре обновления, а в журнале остались записи с номерами 2, 3 и 4. Браузер с курсором 2 получит записи 3 и 4. Курсор 1 тоже пригоден, потому что первая доступная запись следует сразу после него. Курсор 0 потребовал бы потерянное первое обновление. В таком случае сервер отправит событие snapshot с полным текущим состоянием, которое экран может отобразить целиком.

Снимок получает номер последнего включённого обновления. Представьте, что снимок взяли на записи 4, затем появилась запись 5. Клиент после снимка продолжит чтение с 5. Если вместо номера снимка поставить более свежий номер журнала, обновление 5 пропадёт при следующем подключении. В рабочем хранилище читайте состояние и его номер согласованно.

В примере маршрут FastAPI воспроизводит журнал уже завершённой задачи. Поэтому ответ конечный и его можно проверить обычным TestClient. StreamingResponse подходит также для версий FastAPI, где ещё нет fastapi.sse.

Сохраните пример в файл demo.py, установите зависимости командой python -m pip install fastapi httpx и запустите python demo.py.

import json
from collections.abc import Iterator
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient

app = FastAPI()
history = (
    (2, {"status": "validated"}),
    (3, {"status": "packing"}),
    (4, {"status": "ready", "download": "/exports/42"}),
)
current_id, snapshot = history[-1]


def event(seq: int, kind: str, payload: dict[str, str]) -> str:
    data = json.dumps(payload, ensure_ascii=False)
    return f"id: {seq}\nevent: {kind}\ndata: {data}\n\n"


@app.get("/exports/42/events")
def export_events(
    last_event_id: Annotated[int | None, Header(ge=0)] = None,
):
    def replay() -> Iterator[str]:
        if last_event_id is None or not history[0][0] - 1 <= last_event_id <= current_id:
            yield event(current_id, "snapshot", snapshot)
            return
        for seq, update in history:
            if seq > last_event_id:
                yield event(seq, "progress", update)

    return StreamingResponse(replay(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache"})


def parse_events(body: str):
    frames = []
    for block in body.strip().split("\n\n"):
        if block:
            fields = dict(line.split(": ", 1) for line in block.splitlines())
            frames.append((int(fields["id"]), fields["event"], json.loads(fields["data"])))
    return frames


with TestClient(app) as client:
    path = "/exports/42/events"
    replay = client.get(path, headers={"Last-Event-ID": "2"})
    assert replay.headers["content-type"].startswith("text/event-stream")
    assert parse_events(replay.text) == [(3, "progress", history[1][1]), (4, "progress", snapshot)]
    assert [seq for seq, _, _ in parse_events(client.get(path, headers={"Last-Event-ID": "1"}).text)] == [2, 3, 4]
    for expired in ("0", "17"):
        assert parse_events(client.get(path, headers={"Last-Event-ID": expired}).text) == [(4, "snapshot", snapshot)]
    assert parse_events(client.get(path).text) == [(4, "snapshot", snapshot)]
    assert client.get(path, headers={"Last-Event-ID": "-1"}).status_code == 422
    print(parse_events(replay.text))

Клиенту нужны обработчики обоих событий. Подпишитесь через addEventListener("progress", ...) и addEventListener("snapshot", ...). Снимок заменяет состояние экрана целиком. Когда любое из сообщений несёт status: "ready", покажите ссылку и вызовите source.close(). Для завершённой задачи при новом подключении снимок сразу содержит ссылку. Если курсор уже равен 4, экран успел увидеть завершающее событие.

Присоедините живые обновления к тому же журналу

Для незавершённого экспорта оставьте генератор открытым после воспроизведения. Следующую порцию читайте тем же запросом по условию id > cursor ORDER BY id, обновляя курсор по мере отправки. Можно разбудить читающий генератор уведомлением о новой записи. После пробуждения он всё равно дочитывает сам журнал. Так обновление между первоначальной выборкой и подпиской можно забрать при следующей проверке. Для длительных пауз отправляйте комментарии SSE и убедитесь, что прокси пропускает их вовремя.

При нескольких процессах приложения храните журнал и снимок в общей базе. Разрешайте чтение журнала только владельцу задачи, а идентификатор выгрузки берите из проверенного маршрута. На странице, открытой заново, создаётся другой объект EventSource. Снимок даёт ему стартовое состояние и избавляет браузер от ручного восстановления курсора после перезагрузки страницы.

На бесплатном курсе Latorn по FastAPI вы соберёте один сервис с моделями входных данных, авторизацией, базой и тестами маршрутов. Посмотреть программу FastAPI и попробовать открытый урок по TestClient.