29 lines
825 B
Python
29 lines
825 B
Python
import asyncio
|
|
from typing import Dict, Optional
|
|
|
|
|
|
class DeviceTargetCache:
|
|
def __init__(self):
|
|
self.targets: Dict[str, str] = {}
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def get_target(self, device_id: str) -> Optional[str]:
|
|
async with self.lock:
|
|
return self.targets.get(device_id)
|
|
|
|
async def set_target(self, device_id: str, target_device_id: str):
|
|
async with self.lock:
|
|
self.targets[device_id] = target_device_id
|
|
|
|
async def remove_target(self, device_id: str):
|
|
async with self.lock:
|
|
if device_id in self.targets:
|
|
del self.targets[device_id]
|
|
|
|
async def get_all_targets(self) -> list:
|
|
async with self.lock:
|
|
return list(self.targets.items())
|
|
|
|
|
|
# 单例实例
|
|
device_target_cache = DeviceTargetCache() |