Spaces:
Runtime error
Runtime error
File size: 1,864 Bytes
3207814 cb92d2b ff9325e 3207814 cb92d2b 3207814 cb92d2b 3207814 ff9325e 3207814 ff9325e 3207814 ff9325e 3207814 cb92d2b 3207814 cb92d2b 3207814 1d3190d 3207814 be97094 3207814 1d3190d 3207814 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
from typing import Dict
from uuid import UUID
import asyncio
from fastapi import WebSocket
from types import SimpleNamespace
from typing import Dict
from typing import Union
UserDataContent = Dict[UUID, Dict[str, Union[WebSocket, asyncio.Queue]]]
class UserData:
def __init__(self):
self.data_content: Dict[UUID, UserDataContent] = {}
async def create_user(self, user_id: UUID, websocket: WebSocket):
self.data_content[user_id] = {
"websocket": websocket,
"queue": asyncio.Queue(),
}
await asyncio.sleep(1)
def check_user(self, user_id: UUID) -> bool:
return user_id in self.data_content
async def update_data(self, user_id: UUID, new_data: SimpleNamespace):
user_session = self.data_content[user_id]
queue = user_session["queue"]
while not queue.empty():
try:
queue.get_nowait()
except asyncio.QueueEmpty:
continue
await queue.put(new_data)
async def get_latest_data(self, user_id: UUID) -> SimpleNamespace:
user_session = self.data_content[user_id]
queue = user_session["queue"]
try:
return await queue.get()
except asyncio.QueueEmpty:
return None
def delete_user(self, user_id: UUID):
user_session = self.data_content[user_id]
queue = user_session["queue"]
while not queue.empty():
try:
queue.get_nowait()
except asyncio.QueueEmpty:
continue
if user_id in self.data_content:
del self.data_content[user_id]
def get_user_count(self) -> int:
return len(self.data_content)
def get_websocket(self, user_id: UUID) -> WebSocket:
return self.data_content[user_id]["websocket"]
user_data = UserData()
|