Server-Sent Events

python_server_sent_events_server_demo.py

import asyncio
import datetime
import random

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse

app = FastAPI()

app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])


@app.get("/api/sse", summary="test Server-Sent Events")
async def sse_test():
    async def event_generator(n):
        for _ in range(n):
            yield f"data: {_}: {datetime.datetime.now()}\n\n"
            await asyncio.sleep(random.randint(2, 8) / 10)

    return StreamingResponse(event_generator(5), media_type="text/event-stream")


if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8199)