(payload: BaseModel | Dict[str, Any])
| 7853 | |
| 7854 | @staticmethod |
| 7855 | def encode_sse_payload(payload: BaseModel | Dict[str, Any]) -> bytes: |
| 7856 | data = ( |
| 7857 | payload.model_dump(mode="json", exclude_none=True) |
| 7858 | if isinstance(payload, BaseModel) |
| 7859 | else payload |
| 7860 | ) |
| 7861 | return ( |
| 7862 | "data: " |
| 7863 | f"{json.dumps(data, ensure_ascii=False, separators=(',', ':'))}\n\n" |
| 7864 | ).encode("utf-8") |
| 7865 | |
| 7866 | @staticmethod |
| 7867 | def next_stream_chunk( |
no test coverage detected