Два разных уровня границ
Один HTTP-чанк может закончиться посреди UTF-8-последовательности, строки или события. Нельзя передавать каждый чанк напрямую в decode с final=True и нельзя считать его готовым SSE-сообщением. Инкрементальный декодер удерживает неполную последовательность байтов, а отдельный слой накапливает текст до перевода строки. В production-коде следует также определить политику для исключения UnicodeDecodeError: прекратить поток, записать диагностическую причину и не выдавать частично декодированные данные как корректное событие.
import codecs
decoder = codecs.getincrementaldecoder("utf-8")()
for chunk in byte_chunks:
text = decoder.decode(chunk, final=False)
# text передаётся построчному SSE-парсеру
text = decoder.decode(b"", final=True)
# final=True допустим только при подтверждённом завершении байтового потокаМинимальный bounded-парсер
Ниже приведён самостоятельный эскиз, который принимает байтовые чанки, различает CR, LF и CRLF, собирает data-поля и возвращает несколько событий из одного вызова feed. Ограничения заданы в символах уже декодированного текста; это не заменяет отдельный лимит размера входного байтового чанка. Неизвестные поля не превращаются в данные автоматически, а JSON не разбирается внутри парсера.
import codecs
class ParseLimitError(ValueError):
pass
class IncompleteStream(ValueError):
pass
class SSEParser:
def __init__(self, *, max_line_chars=8192, max_event_chars=1_048_576):
self._decoder = codecs.getincrementaldecoder("utf-8")()
self._line = []
self._first_char = True
self._pending_cr = False
self._data = []
self._event = None
self._max_line_chars = max_line_chars
self._max_event_chars = max_event_chars
self._event_chars = 0
def feed(self, chunk):
text = self._decoder.decode(chunk, final=False)
events = []
self._consume_text(text, events)
return events
def close(self):
events = []
text = self._decoder.decode(b"", final=True)
self._consume_text(text, events)
if self._line:
raise IncompleteStream("поток завершился внутри строки")
if self._data or self._event is not None:
raise IncompleteStream("поток завершился до пустой строки")
return events
def _consume_text(self, text, events):
for char in text:
if self._first_char:
self._first_char = False
if char == "\ufeff":
continue
if self._pending_cr:
self._pending_cr = False
if char == "\n":
continue
if char == "\r":
self._emit_line(events)
self._pending_cr = True
elif char == "\n":
self._emit_line(events)
else:
self._line.append(char)
if len(self._line) > self._max_line_chars:
raise ParseLimitError("превышен лимит длины строки")
def _emit_line(self, events):
line = "".join(self._line)
self._line.clear()
if line == "":
if self._data:
events.append({
"event": self._event or "message",
"data": "\n".join(self._data),
})
self._data.clear()
self._event = None
self._event_chars = 0
return
if line.startswith(":"):
return
field, separator, value = line.partition(":")
if not separator:
value = ""
elif value.startswith(" "):
value = value[1:]
if field == "data":
projected = self._event_chars + len(value)
if self._data:
projected += 1
if projected > self._max_event_chars:
raise ParseLimitError("превышен лимит размера события")
self._data.append(value)
self._event_chars = projected
elif field == "event":
self._event = valueСемантика полей и границы события
Пустая строка завершает текущую запись. Несколько строк вида data: образуют одно поле data с переводами строки между значениями. Строка, начинающаяся с двоеточия, является комментарием и не должна попадать в payload. В примере поле event сохраняется как имя события, но остальные поля намеренно не реализованы: их следует добавлять только при наличии явного требования протокола. Если запись не завершена пустой строкой, close() сообщает об этом, а не выдаёт частичный результат. Это ограниченный учебный профиль, а не полная реализация EventSource: id/retry не поддерживаются, некорректный UTF-8 отклоняется. При EOF стандарт SSE отбрасывает недописанное событие; здесь намеренно выдаётся IncompleteStream, чтобы локальная проверка явно обнаружила обрыв. Завершённая CR строка не считается обрывом.
raw = "data: первая строка\r\ndata: кириллица\r\n\r\n".encode("utf-8")
parser = SSEParser()
events = parser.feed(raw[:18])
events += parser.feed(raw[18:])
events += parser.close()
# Ожидаемая структура результата по смыслу:
# [{"event": "message", "data": "первая строка\nкириллица"}]
# Этот фрагмент является примером сценария, а не заявлением о результате запуска.Не смешивать SSE и JSON
Собранное data-поле может содержать JSON, текст, служебный маркер или другой формат. Сам факт наличия фигурной скобки не делает фрагмент готовым JSON-ответом. Сначала нужно применить правила SSE, затем проверить контракт payload: допустимую схему, обязательные поля, размер и тип события. Маркер окончания нельзя считать универсальным: его значение и формат зависят от источника. При ошибке JSON фиксируйте только разрешённую метку причины и номер синтетического fixture; не записывайте payload, токены или полные тела ответов. Не продолжайте обработку, как будто событие было валидным.
import json
def parse_json_payload(event):
if event["event"] != "message":
return None
try:
return json.loads(event["data"])
except json.JSONDecodeError as exc:
raise ValueError("data не соответствует ожидаемому JSON-контракту") from excОфлайн-fixtures и явно проверяемые ограничения
Набор fixtures должен проверять parser, а не сетевой клиент или конкретную модель. Нужны как минимум такие случаи: многобайтный кириллический символ, разделённый между чтениями; CRLF, разделённый между чанками; несколько data-строк; комментарий; две записи в одном чтении; пустая запись; незавершённая строка в конце; превышение лимита строки и события; некорректная UTF-8-последовательность; JSON, разбитый между SSE-строками. Для каждого fixture следует отдельно фиксировать ожидаемое событие или ожидаемое исключение. Эти проверки не доказывают корректность конкретной HTTP-библиотеки, прокси, сервера или production streaming.
FIXTURES = {
"two_events_one_read": (
b": keepalive\r\n"
b"data: {\"part\": 1}\r\n\r\n"
b"data: {\"part\": 2}\r\n\r\n"
),
"split_utf8": [
"data: кириллица\n\n".encode("utf-8")[:8],
"data: кириллица\n\n".encode("utf-8")[8:],
],
"split_crlf": [b"data: x\r", b"\n\r", b"\n"],
"unterminated_event": b"data: incomplete\n",
}
# Для каждого fixture нужно написать отдельную проверку.
# Запуск и результаты здесь не заявляются.Шаблон промпта
Написать на русском оригинальный инженерный гайд о bounded SSE-парсере в Python без сети и без вызова модели. Разделить инкрементальное декодирование UTF-8, построчную обработку SSE и последующий разбор payload. Объяснить, почему транспортные чанки не являются границами символов или событий; показать накопление data-полей, комментарии, CR/LF/CRLF и пустую строку как границу события. Привести эскиз кода с лимитами строки и события, явным поведением при переполнении, некорректной UTF-8 и незавершённом потоке. Добавить офлайн-fixtures для разрыва кириллического символа, разрыва CRLF, нескольких data-строк, комментариев, нескольких событий в одном чтении и незавершённой записи. Не выдавать JSON-фрагмент за готовый ответ модели, не предполагать универсальный маркер окончания и не обещать поддержку конкретного провайдера или модели. Не утверждать, что код запускался или тесты пройдены.