sse sub resource

This commit is contained in:
2026-07-12 00:51:23 +05:30
parent 4a7a76e330
commit f1a7a556fd
5 changed files with 59 additions and 60 deletions

View File

@@ -24,9 +24,8 @@ library API surface.
OpenAPI x- extension fields demonstrated
----------------------------------------------------------------------
Schema-level extensions (mark a schema as a UI resource):
Schema-level extensions (display metadata for resource endpoints):
``x-resource`` (REQUIRED) Maps schema to URL path segment
``x-primary-key`` (REQUIRED) Primary key property name
``x-display-format`` (REQUIRED) Human-readable label template
``x-list-columns`` (REQUIRED) Columns for the datatable

View File

@@ -31,16 +31,12 @@ 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()
pass
app = OpenAPIFirstApp(

View File

@@ -60,7 +60,6 @@ components:
Call:
type: object
x-resource: calls
x-primary-key: _received_at
x-display-format: "{sound}"
x-list-columns: [sound]
@@ -75,7 +74,6 @@ components:
Parent:
type: object
x-resource: parents
x-primary-key: id
x-display-format: "{name}"
x-list-columns: [name, email, phone]
@@ -114,7 +112,6 @@ components:
Vet:
type: object
x-resource: vets
x-primary-key: id
x-display-format: "Dr. {name}"
x-list-columns: [name, specialty, email, phone]
@@ -158,7 +155,6 @@ components:
Treatment:
type: object
x-resource: treatments
x-primary-key: id
x-display-format: "{label}"
x-list-columns: [label, description]
@@ -189,7 +185,6 @@ components:
Pet:
type: object
x-resource: pets
x-primary-key: id
x-display-format: "{name} #{id}"
x-list-columns: [name, species, age, weight, birthDate, parents]
@@ -260,7 +255,6 @@ components:
Appointment:
type: object
x-resource: appointments
x-primary-key: id
x-display-format: "Appt #{id} {date}"
x-list-columns: [date, pet, vet, treatment, notes]
@@ -295,7 +289,7 @@ components:
$ref: '#/components/schemas/Pet'
x-fk:
resource: pets
x-order: 3
x-order: 4
x-label: "Pet"
x-description: "Select a pet"
x-filterable: true
@@ -304,7 +298,7 @@ components:
x-fk:
resource: vets
prefetch: true
x-order: 4
x-order: 5
x-label: "Veterinarian"
x-description: "Select a veterinarian"
x-filterable: true
@@ -313,13 +307,13 @@ components:
x-fk:
resource: treatments
prefetch: true
x-order: 5
x-order: 6
x-label: "Treatment"
x-description: "Select a treatment"
x-filterable: true
metadata:
$ref: '#/components/schemas/Metadata'
x-order: 4
x-order: 7
x-label: "Metadata"
required: [id, date, pet, vet, treatment]
@@ -384,19 +378,6 @@ components:
$ref: '#/components/schemas/ErrorBody'
paths:
/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:
$ref: '#/components/schemas/Call'
/parents:
get:
summary: List parents (paginated)
@@ -945,6 +926,23 @@ paths:
$ref: '#/components/responses/ValidationError'
'500':
$ref: '#/components/responses/InternalServerError'
/pets/{id}/calls:
get:
summary: Stream animal sounds via SSE, scoped to a pet's species
operationId: stream_calls
x-sse: true
parameters:
- name: id
in: path
required: true
schema: {type: integer}
responses:
'200':
description: SSE stream of random animal sounds
content:
text/event-stream:
schema:
$ref: '#/components/schemas/Call'
/appointments:
get:

View File

@@ -43,6 +43,7 @@ from data import (
create_pet as _create_pet,
update_pet as _update_pet,
delete_pet as _delete_pet,
get_pet as _get_pet,
list_appointments as _list_appointments,
get_appointment as _get_appointment,
create_appointment as _create_appointment,
@@ -368,9 +369,14 @@ def delete_appointment(id: int, response: Response):
response.status_code = 204
async def stream_calls():
"""Stream random animal sounds via SSE."""
q = await subscribe()
async def stream_calls(id: int):
"""Stream animal sounds via SSE, scoped to a pet's species."""
try:
pet = _get_pet(id)
except KeyError:
raise HTTPException(status_code=404, detail="Pet not found")
species = pet.species
q = await subscribe(id, species)
async def event_generator():
try:
@@ -378,6 +384,6 @@ async def stream_calls():
data = await q.get()
yield f"data: {data}\n\n"
finally:
unsubscribe(q)
unsubscribe(id, q)
return StreamingResponse(event_generator(), media_type="text/event-stream")

View File

@@ -1,44 +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
_sounds_by_species = {"dog": ["woof"], "cat": ["meow"], "bird": ["coo"]}
_subscribers: dict[int, list[asyncio.Queue]] = {}
_worker_tasks: dict[int, asyncio.Task] = {}
async def _sound_worker():
async def _sound_worker(pet_id: int, species: str):
sounds = _sounds_by_species.get(species, ["woof"])
while True:
sound = random.choice(_sounds)
sound = random.choice(sounds)
data = json.dumps({"sound": sound})
for q in _subscribers:
queues = _subscribers.get(pet_id, [])
for q in queues:
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 _ensure_worker(pet_id: int, species: str):
if pet_id not in _worker_tasks or _worker_tasks[pet_id].done():
_subscribers.setdefault(pet_id, [])
_worker_tasks[pet_id] = asyncio.create_task(
_sound_worker(pet_id, species)
)
def stop_worker():
if _worker_task is not None:
_worker_task.cancel()
async def subscribe() -> asyncio.Queue:
async def subscribe(pet_id: int, species: str) -> asyncio.Queue:
q: asyncio.Queue = asyncio.Queue()
_subscribers.append(q)
_subscribers.setdefault(pet_id, []).append(q)
_ensure_worker(pet_id, species)
return q
def unsubscribe(q: asyncio.Queue):
if q in _subscribers:
_subscribers.remove(q)
def unsubscribe(pet_id: int, q: asyncio.Queue):
queues = _subscribers.get(pet_id, [])
if q in queues:
queues.remove(q)
if not queues:
task = _worker_tasks.pop(pet_id, None)
if task and not task.done():
task.cancel()
_subscribers.pop(pet_id, None)