Pentru a construi o cameră de chat, un tablou de bord colaborativ sau un sistem de alerte în timp real, serverul trebuie să trimită un mesaj către toți clienții conectați simultan. Acest lucru se realizează prin pattern-ul ConnectionManager.
Managerul menține o listă internă de instanțe WebSocket active, gestionează conectarea/deconectarea lor și iterează peste listă pentru a difuza mesajul (broadcast).
Exemplu: Implementare ConnectionManager completă pentru camere de chat
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
app = FastAPI()
@app.websocket("/ws/chat")
async def chat_ws(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Mesaj public: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("Un utilizator a părăsit conversația.")
Întrebări & Exemple de Cod:
Pattern-ul ConnectionManager pentru WebSocketsAi o nelămurire despre utilizarea comenzii
Pattern-ul ConnectionManager pentru WebSocketssau ai descoperit un truc / fragment de cod util? Distribuie-l cu ceilalți programatori!Fii primul care pune o întrebare sau distribuie un fragment de cod / prompt!
Nu au fost adăugate încă întrebări sau exemple de cod pentru această pagină. Dacă ai întâlnit o problemă sau ai un snippet util, adaugă-l mai jos pentru a-i ajuta pe ceilalți.