patx/micropie
make server sent events example to main examples dir
Commit 29450c9 · patx · 2025-07-09T17:17:46-04:00
Comments
No comments yet.
Diff
diff --git a/examples/server_sent_events/app.py b/examples/server_sent_events/app.py
new file mode 100644
index 0000000..bd469f0
--- /dev/null
+++ b/examples/server_sent_events/app.py
@@ -0,0 +1,55 @@
+import asyncio
+import random
+from micropie import App
+
+
+class MyApp(App):
+
+ async def index(self):
+ return """
+ <!DOCTYPE html>
+ <html>
+ <head>
+ <title>Server-Sent Events Demo with MicroPie</title>
+ </head>
+ <body>
+ <h1>Server-Sent Events with MicroPie</h1>
+ <div id="output"></div>
+ <script>
+ const source = new EventSource('/events');
+ source.onmessage = function(event) {
+ const newElement = document.createElement("p");
+ newElement.textContent = `Received: ${event.data}`;
+ document.getElementById("output").appendChild(newElement);
+ };
+ source.onerror = function() {
+ console.log("SSE error occurred");
+ };
+ </script>
+ </body>
+ </html>
+ """
+
+ # 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)
diff --git a/examples/server_sent_events/chat.py b/examples/server_sent_events/chat.py
new file mode 100644
index 0000000..5a47192
--- /dev/null
+++ b/examples/server_sent_events/chat.py
@@ -0,0 +1,91 @@
+from micropie import App
+from collections import deque
+import json
+import asyncio
+
+class ChatApp(App):
+ def __init__(self):
+ super().__init__()
+ self.messages = deque(maxlen=100)
+ self.clients = set()
+
+ async def index(self):
+ # Inline HTML for demo purposes
+ html = """
+ <!doctype html>
+ <title>Micropie Chat SSE</title>
+ <h1>Micropie Chat SSE Demo</h1>
+ <form id="chat-form">
+ Name: <input id="username" value="Anonymous" size="10">
+ <input id="message" autocomplete="off" size="40">
+ <button>Send</button>
+ </form>
+ <pre id="log"></pre>
+ <script>
+ const log = document.getElementById('log');
+ const usernameInput = document.getElementById('username');
+ const messageInput = document.getElementById('message');
+ const form = document.getElementById('chat-form');
+
+ const evtSource = new EventSource('/events');
+ evtSource.onmessage = function(event) {
+ const data = JSON.parse(event.data);
+ log.textContent += data.username + ": " + data.message + "\\n";
+ };
+
+ form.onsubmit = async function(e) {
+ e.preventDefault();
+ await fetch('/send', {
+ method: 'POST',
+ headers: {'Content-Type': 'application/json'},
+ body: JSON.stringify({
+ username: usernameInput.value,
+ message: messageInput.value,
+ })
+ });
+ messageInput.value = "";
+ };
+ </script>
+ """
+ return 200, html, [('Content-Type', 'text/html')]
+
+ async def send(self):
+ data = self.request.get_json
+ username = data.get("username", "Anonymous")
+ message = data.get("message", "")
+ if message.strip():
+ msg = {"username": username, "message": message}
+ self.messages.append(msg)
+ # Broadcast to all connected clients
+ for client in self.clients.copy():
+ try:
+ await client.put(json.dumps(msg))
+ except Exception:
+ self.clients.discard(client)
+ return 200, {"status": "success"}
+
+ async def events(self):
+ async def stream():
+ queue = asyncio.Queue()
+ setattr(stream, "_micropie_disconnect_queue", queue)
+ self.clients.add(queue)
+ try:
+ for msg in self.messages:
+ yield f"data: {json.dumps(msg)}\n\n"
+ while True:
+ try:
+ msg = await queue.get()
+ if msg is None:
+ break
+ yield f"data: {msg}\n\n"
+ except asyncio.CancelledError:
+ break
+ finally:
+ self.clients.discard(queue)
+ return 200, stream(), [
+ ('Content-Type', 'text/event-stream'),
+ ('Cache-Control', 'no-cache'),
+ ('Connection', 'keep-alive')
+ ]
+
+app = ChatApp()