| 12345678910111213141516171819202122232425262728293031323334353637383940414243 |
- import asyncio
- import json
- from fastapi import Request, Response
- from fastapi.routing import APIRoute
- from app.core.events import broadcast
- class SSEBroadcasterRoute(APIRoute):
- def get_route_handler(self):
- original_route_handler = super().get_route_handler()
- async def custom_route_handler(request: Request) -> Response:
- # 1. Execute the route and let FastAPI serialize the response
- response = await original_route_handler(request)
- openapi_extra = self.openapi_extra or {}
- event_name = openapi_extra.get("sse_event")
- # 2. Check if an event is declared and the request was successful
- if event_name and response.status_code in (200, 201):
- project_id = request.path_params.get("project_id")
- client_id = request.headers.get("X-Client-ID", "unknown")
- if request.method == "DELETE":
- # For DELETE, the response body is often empty. We broadcast the path parameters.
- item_json = json.dumps(request.path_params)
- else:
- # For POST/PUT, response.body is already serialized as a JSON string by FastAPI
- item_json = response.body.decode("utf-8")
- # Assemble the raw payload string
- payload_str = (
- f'{{"client_id": "{client_id}", "event": "{event_name}", "item": {item_json}}}'
- )
- channel = f"project_{project_id}"
- asyncio.create_task(broadcast.publish(channel, payload_str))
- return response
- return custom_route_handler
|