Python / FastAPI Basics Interview Questions
How do you implement WebSocket endpoints in FastAPI?
FastAPI supports WebSockets natively via the WebSocket parameter type. Use await websocket.accept() to establish the connection, then loop to send and receive messages.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect from typing import List app = FastAPI() # Simple echo WebSocket @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() try: while True: data = await websocket.receive_text() await websocket.send_text(f"Echo: {data}") except WebSocketDisconnect: print("Client disconnected") # Connection manager for broadcast (e.g. chat room) class ConnectionManager: def __init__(self): self.active: List[WebSocket] = [] async def connect(self, ws: WebSocket): await ws.accept() self.active.append(ws) def disconnect(self, ws: WebSocket): self.active.remove(ws) async def broadcast(self, message: str): for ws in self.active: await ws.send_text(message) manager = ConnectionManager() @app.websocket("/chat/{room}") async def chat(websocket: WebSocket, room: str): await manager.connect(websocket) try: while True: msg = await websocket.receive_text() await manager.broadcast(f"[{room}] {msg}") except WebSocketDisconnect: manager.disconnect(websocket) await manager.broadcast(f"A user left {room}")
WebSocket message types: receive_text() / send_text() for strings, receive_bytes() / send_bytes() for binary, receive_json() / send_json() for JSON.
More Related questions...