import asyncio
import random
from micropie import App
class MyApp(App):
async def index(self):
return """
Server-Sent Events Demo with MicroPie
Server-Sent Events with MicroPie
"""
# SSE endpoint to stream random numbers
async def events(self):
async def event_generator():
try:
while True:
# Generate random number
data = f"Random number: {random.randint(1, 100)}"
# Yield SSE formatted data
yield f"data: {data}\n\n".encode("utf-8")
# Wait for 1 second before sending next event
await asyncio.sleep(1)
except asyncio.CancelledError:
print("Client disconnected")
pass
# Return status code, async generator, and headers for SSE
return 200, event_generator(), [("Content-Type", "text/event-stream")]
app = MyApp()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)