"""
Minimal Server-Sent Events (SSE) example demonstrating core concepts.
This example shows the essential SSE patterns:
- Simple event streaming with automatic termination
- Endless event streaming with proper cleanup
- HTML client with JavaScript EventSource
- Both finite and infinite stream patterns
For more advanced features, see the examples/ directory.
Usage:
python example.py
Then visit: http://localhost:8000
"""
import asyncio
import uvicorn
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import HTMLResponse
from starlette.routing import Route
from sse_starlette import EventSourceResponse
HTML_CLIENT = """
SSE Example
Server-Sent Events Example
Finite Stream (numbers 1-5)
Click start to begin...
Endless Stream
Click start to begin...
"""
async def finite_number_stream():
"""
Demonstrates finite SSE stream that automatically terminates.
This is the simplest SSE pattern - stream a fixed set of data
and let the connection close naturally when the generator ends.
"""
for number in range(1, 6): # Numbers 1 through 5
# Each yield sends one SSE event to the client
# The dict format automatically creates: data: {number}
yield {"data": number}
await asyncio.sleep(1) # 1 second between events
async def endless_counter_stream(request: Request):
"""
Demonstrates endless SSE stream with proper cleanup.
Key concepts:
- Infinite loop for continuous streaming
- Client disconnect detection via request.is_disconnected()
- Exception handling for graceful cleanup
- asyncio.CancelledError for proper resource cleanup
"""
counter = 0
try:
while True:
counter += 1
yield {"data": counter}
await asyncio.sleep(0.5) # 500ms between events
except asyncio.CancelledError:
# This exception is raised when:
# 1. Client closes the browser/tab
# 2. Client calls eventSource.close()
# 3. Server is shutting down
#
# Always re-raise CancelledError to ensure proper cleanup
raise
# Starlette route handlers
async def home_page(request: Request) -> HTMLResponse:
"""Serve the HTML demo page."""
return HTMLResponse(HTML_CLIENT)
async def numbers_endpoint(request: Request) -> EventSourceResponse:
"""SSE endpoint for finite number stream."""
return EventSourceResponse(finite_number_stream())
async def endless_endpoint(request: Request) -> EventSourceResponse:
"""SSE endpoint for endless counter stream."""
return EventSourceResponse(endless_counter_stream(request))
# Application setup
app = Starlette(
routes=[
Route("/", endpoint=home_page),
Route("/numbers", endpoint=numbers_endpoint),
Route("/endless", endpoint=endless_endpoint),
]
)
if __name__ == "__main__":
print("Starting SSE example server...")
print("Visit: http://localhost:8000")
print("\nEndpoints:")
print(" / - Demo page with JavaScript client")
print(" /numbers - Finite stream (1-5)")
print(" /endless - Infinite counter stream")
uvicorn.run(app, host="0.0.0.0", port=8000)