52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
"""
|
|
SSE endpoint — clients connect here to receive live task progress.
|
|
|
|
GET /api/v1/events?task_id=<uuid>
|
|
|
|
The connection stays open; the server pushes events as the task runs.
|
|
Nginx must have `proxy_buffering off` for this to work through a reverse proxy.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from app.dependencies import get_current_user
|
|
from app.models.user import User
|
|
from app.services.sse import sse_manager
|
|
|
|
router = APIRouter(prefix="/api/v1/events", tags=["events"])
|
|
|
|
|
|
@router.get("")
|
|
async def event_stream(
|
|
request: Request,
|
|
task_id: str,
|
|
_: User = Depends(get_current_user),
|
|
):
|
|
queue: asyncio.Queue = asyncio.Queue(maxsize=200)
|
|
await sse_manager.subscribe(task_id, queue)
|
|
|
|
async def generator():
|
|
try:
|
|
while True:
|
|
if await request.is_disconnected():
|
|
break
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout=25.0)
|
|
yield event.encode()
|
|
except asyncio.TimeoutError:
|
|
yield ": keepalive\n\n"
|
|
finally:
|
|
await sse_manager.unsubscribe(task_id, queue)
|
|
|
|
return StreamingResponse(
|
|
generator(),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"X-Accel-Buffering": "no", # tells Nginx to disable buffering
|
|
"Connection": "keep-alive",
|
|
},
|
|
)
|