30 lines
892 B
Python
30 lines
892 B
Python
import asyncio
|
|
from typing import Dict, Optional
|
|
from fastapi import WebSocket
|
|
|
|
|
|
class ConnectionManager:
|
|
def __init__(self):
|
|
self.connections: Dict[str, WebSocket] = {}
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def add_connection(self, device_id: str, websocket: WebSocket):
|
|
async with self.lock:
|
|
self.connections[device_id] = websocket
|
|
|
|
async def remove_connection(self, device_id: str):
|
|
async with self.lock:
|
|
if device_id in self.connections:
|
|
del self.connections[device_id]
|
|
|
|
async def get_connection(self, device_id: str) -> Optional[WebSocket]:
|
|
async with self.lock:
|
|
return self.connections.get(device_id)
|
|
|
|
async def get_all_connections(self) -> Dict[str, WebSocket]:
|
|
async with self.lock:
|
|
return dict(self.connections)
|
|
|
|
|
|
connection_manager = ConnectionManager()
|