eksplicit mapping af envs
This commit is contained in:
25
backend/app/services/ws_manager.py
Normal file
25
backend/app/services/ws_manager.py
Normal file
@@ -0,0 +1,25 @@
|
||||
import json
|
||||
|
||||
from fastapi import WebSocket
|
||||
|
||||
|
||||
class ConnectionManager:
|
||||
"""Manages active WebSocket connections keyed by BLE token."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._connections: dict[str, WebSocket] = {}
|
||||
|
||||
async def connect(self, token: str, websocket: WebSocket) -> None:
|
||||
await websocket.accept()
|
||||
self._connections[token] = websocket
|
||||
|
||||
def disconnect(self, token: str) -> None:
|
||||
self._connections.pop(token, None)
|
||||
|
||||
async def send(self, token: str, message: dict) -> None:
|
||||
websocket = self._connections.get(token)
|
||||
if websocket is not None:
|
||||
await websocket.send_text(json.dumps(message))
|
||||
|
||||
|
||||
manager = ConnectionManager()
|
||||
Reference in New Issue
Block a user