starlette_sse_server_sent_events_streaming_endpoint.py

1import asyncio
2import uvicorn
3from starlette.applications import Starlette
4from starlette.routing import Route
5from sse_starlette.sse import EventSourceResponse
6
7async def numbers(minimum, maximum):
8    for i in range(minimum, maximum + 1):
9        await asyncio.sleep(0.9)
10        yield {"data": i}
11
12async def sse(request):
13    generator = numbers(1, 5)
14    return EventSourceResponse(generator)
15
16routes = [
17    Route("/", endpoint=sse)
18]
19
20app = Starlette(debug=True, routes=routes)
21
22if __name__ == "__main__":
23    uvicorn.run(app, host="0.0.0.0", port=8000)