FastAPI WebSockets — Real-time Communication

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Dict

app = FastAPI()

class ForgeSocketManager:
    """Manages active WebSocket connections with unique identifiers."""
    def __init__(self):
        # Map user IDs to their respective WebSocket objects
        self.active_connections: Dict[str, WebSocket] = {}

    async def connect(self, user_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active_connections[user_id] = websocket

    def disconnect(self, user_id: str):
        if user_id in self.active_connections:
            del self.active_connections[user_id]

    async def send_personal_message(self, message: str, user_id: str):
        if user_id in self.active_connections:
            await self.active_connections[user_id].send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections.values():
            await connection.send_text(message)

manager = ForgeSocketManager()

@app.websocket('/ws/{user_id}')
async def websocket_endpoint(websocket: WebSocket, user_id: str):
    await manager.connect(user_id, websocket)
    try:
        while True:
            # Listen for incoming messages
            data = await websocket.receive_text()
            await manager.broadcast(f'User {user_id} sent: {data}')
    except WebSocketDisconnect:
        manager.disconnect(user_id)
        await manager.broadcast(f'User {user_id} has left the forge.')

from fastapi import FastAPI, WebSocket, Query, status

app = FastAPI()

async def get_token_header(token: str = Query(...)):
    # In production, verify this JWT against your Auth provider
    if token != "secret-forge-token":
        return False
    return True

@app.websocket('/secure-ws')
async def secure_socket(websocket: WebSocket, token: str = Query(...)):
    is_valid = await get_token_header(token)
    if not is_valid:
        # Close with policy violation code 1008 if unauthorized
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    
    await websocket.accept()
    await websocket.send_text('Access Granted. Welcome to the Forge.')
    await websocket.close()