fastapi_server_sent_events_streaming_response_generator.py

1import asyncio
2from fastapi import FastAPI
3from fastapi.responses import StreamingResponse
4
5app = FastAPI()
6
7async def fake_video_streamer():
8    for i in range(10):
9        yield f"data: Message {i}\n\n"
10        await asyncio.sleep(1)
11
12@app.get("/")
13async def main():
14    return StreamingResponse(fake_video_streamer(), media_type="text/event-stream")