sse events sample

openapi_first/templates/vet_app/main.py

@@ -25,15 +25,29 @@ Example:
     uvicorn main:app
 """
 
+from contextlib import asynccontextmanager
+
 from starlette.middleware.cors import CORSMiddleware
 
 from openapi_first.app import OpenAPIFirstApp
 import routes
+from sse import start_worker, stop_worker
+
+
+@asynccontextmanager
+async def lifespan(app):
+    start_worker()
+    try:
+        yield
+    finally:
+        stop_worker()
+
 
 app = OpenAPIFirstApp(
     openapi_path="openapi.yaml",
     routes_module=routes,
     title="Veterinary Clinic Service",
+    lifespan=lifespan,
 )
 
 app.add_middleware(

openapi_first/templates/vet_app/openapi.yaml

@@ -984,6 +984,25 @@ paths:
           $ref: '#/components/responses/ValidationError'
         '500':
           $ref: '#/components/responses/InternalServerError'
+  /calls:
+    get:
+      summary: Stream random animal sounds via SSE
+      operationId: stream_calls
+      x-sse: true
+      responses:
+        '200':
+          description: SSE stream of random animal sounds
+          content:
+            text/event-stream:
+              schema:
+                type: object
+                required: [sound]
+                properties:
+                  sound:
+                    type: string
+                    enum: [woof, meow, coo]
+                    description: Random animal sound
+
   /appointments/{id}:
     get:
       operationId: get_appointment

openapi_first/templates/vet_app/routes.py

@@ -11,6 +11,9 @@ in the OpenAPI specification.
 """
 
 from fastapi import Response, HTTPException, UploadFile
+from fastapi.responses import StreamingResponse
+
+from sse import subscribe, unsubscribe
 
 from models import (
     ParentCreate,
@@ -363,3 +366,18 @@ def delete_appointment(id: int, response: Response):
     except KeyError:
         raise HTTPException(status_code=404, detail="Appointment not found")
     response.status_code = 204
+
+
+async def stream_calls():
+    """Stream random animal sounds via SSE."""
+    q = await subscribe()
+
+    async def event_generator():
+        try:
+            while True:
+                data = await q.get()
+                yield f"data: {data}\n\n"
+        finally:
+            unsubscribe(q)
+
+    return StreamingResponse(event_generator(), media_type="text/event-stream")

openapi_first/templates/vet_app/sse.py

@@ -0,0 +1,44 @@
+"""
+SSE broadcast for the animal-sounds worker.
+
+Not part of the openapi_first library API surface.
+"""
+
+import asyncio
+import random
+import json
+
+
+_sounds = ["woof", "meow", "coo"]
+_subscribers: list[asyncio.Queue] = []
+_worker_task: asyncio.Task | None = None
+
+
+async def _sound_worker():
+    while True:
+        sound = random.choice(_sounds)
+        data = json.dumps({"sound": sound})
+        for q in _subscribers:
+            await q.put(data)
+        await asyncio.sleep(random.uniform(1, 5))
+
+
+def start_worker():
+    global _worker_task
+    _worker_task = asyncio.create_task(_sound_worker())
+
+
+def stop_worker():
+    if _worker_task is not None:
+        _worker_task.cancel()
+
+
+async def subscribe() -> asyncio.Queue:
+    q: asyncio.Queue = asyncio.Queue()
+    _subscribers.append(q)
+    return q
+
+
+def unsubscribe(q: asyncio.Queue):
+    if q in _subscribers:
+        _subscribers.remove(q)