from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request
from fastapi.responses import HTMLResponse
import asyncio, json
from typing import Set
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active: Set[WebSocket] = set()
self._lock = asyncio.Lock()
async def connect(self, ws: WebSocket):
await ws.accept()
async with self._lock:
self.active.add(ws)
async def disconnect(self, ws: WebSocket):
async with self._lock:
self.active.discard(ws)
async def broadcast(self, message: str):
data = json.dumps({"text": message})
async with self._lock:
dead = []
for ws in self.active:
try:
await ws.send_text(data)
except Exception:
dead.append(ws)
for ws in dead:
self.active.discard(ws)
manager = ConnectionManager()
@app.get("/", response_class=HTMLResponse)
async def index():
return """
Messages
"""
@app.post("/send")
async def send(request: Request):
text = ""
try:
payload = await request.json()
if isinstance(payload, dict):
text = payload.get("text", "")
except Exception:
try:
form = await request.form()
text = form.get("text", "")
except Exception:
pass
await manager.broadcast(text or "")
return {"message": text or ""}
@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
await manager.connect(ws)
try:
while True:
await ws.receive_text()
except WebSocketDisconnect:
await manager.disconnect(ws)