6b50ffba6a
Use FastAPI's native `StreamingResponse` for less dependencies. This initially didn't work because I didn't include the additional header specifying the content type, which is an event stream. ref: N25B-110
39 lines
1023 B
Python
39 lines
1023 B
Python
import asyncio
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel
|
|
import datetime
|
|
|
|
class Message(BaseModel):
|
|
message: str
|
|
|
|
app = FastAPI()
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["http://localhost:5173"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@app.post("/message")
|
|
async def receive_message(message: Message):
|
|
print(f"Received message: {message}")
|
|
return { "status": "Message received" }
|
|
|
|
@app.get("/sse")
|
|
async def sse_endpoint(request: Request):
|
|
async def event_generator():
|
|
while True:
|
|
if await request.is_disconnected():
|
|
break
|
|
|
|
current_time = datetime.datetime.now().strftime("%H:%M:%S")
|
|
yield f"data: Server time: {current_time}\n\n"
|
|
await asyncio.sleep(1)
|
|
|
|
return StreamingResponse(event_generator(), media_type="text/event-stream")
|