sse_parser.py (1377B)
1 """SSE (Server-Sent Events) stream parser.""" 2 3 from __future__ import annotations 4 5 import json 6 from collections.abc import Generator 7 8 9 def _parse_data_json(data_line: str) -> dict | None: # pragma: no mutate: block 10 """Return parsed JSON from a ``data:`` line, or ``None`` on failure.""" 11 raw = data_line.removeprefix("data:").strip() 12 if not raw: 13 return None 14 try: 15 return json.loads(raw) 16 except json.JSONDecodeError: 17 return None 18 19 20 def _parse_event_type(line: str) -> str | None: # pragma: no mutate: block 21 """Return the event type from an ``event:`` line, or ``None``.""" 22 stripped = line.removeprefix("event:").strip() 23 return stripped or None 24 25 26 def parse_sse_events(text: str) -> Generator[dict, None, None]: # pragma: no mutate: block 27 """Yield event dicts from an SSE text stream.""" 28 current_event: str | None = None 29 current_data: dict | None = None 30 31 for line in text.split("\n"): 32 if line.startswith(":"): 33 continue 34 if line.startswith("event:"): 35 current_event = _parse_event_type(line) 36 elif line.startswith("data:"): 37 current_data = _parse_data_json(line) 38 elif line == "" and current_event and current_data: 39 yield {"event": current_event, "data": current_data} 40 current_event = None 41 current_data = None