14 lines
370 B
Python
14 lines
370 B
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
|
|
def event(name: str, data: dict[str, Any]) -> bytes:
|
|
return f"event: {name}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode("utf-8")
|
|
|
|
|
|
async def one_event(name: str, data: dict[str, Any]) -> AsyncIterator[bytes]:
|
|
yield event(name, data)
|