66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
import asyncio
|
|
from uuid import UUID, uuid4
|
|
|
|
from app.models import CurrentStatus, Monitor, MonitorCreate, MonitorUpdate, utc_now
|
|
|
|
|
|
class MonitorNotFound(KeyError):
|
|
pass
|
|
|
|
|
|
class MonitorStore:
|
|
"""Copy-in/copy-out process-local store guarded by one asyncio lock."""
|
|
|
|
def __init__(self) -> None:
|
|
self._items: dict[UUID, Monitor] = {}
|
|
self._lock = asyncio.Lock()
|
|
|
|
async def create(self, data: MonitorCreate) -> Monitor:
|
|
now = utc_now()
|
|
monitor = Monitor(
|
|
id=uuid4(),
|
|
created_at=now,
|
|
updated_at=now,
|
|
current_status=CurrentStatus(),
|
|
**data.model_dump(),
|
|
)
|
|
async with self._lock:
|
|
self._items[monitor.id] = monitor
|
|
return monitor.model_copy(deep=True)
|
|
|
|
async def list(self) -> list[Monitor]:
|
|
async with self._lock:
|
|
values = sorted(self._items.values(), key=lambda item: item.created_at)
|
|
return [item.model_copy(deep=True) for item in values]
|
|
|
|
async def get(self, monitor_id: UUID) -> Monitor:
|
|
async with self._lock:
|
|
item = self._items.get(monitor_id)
|
|
if item is None:
|
|
raise MonitorNotFound(monitor_id)
|
|
return item.model_copy(deep=True)
|
|
|
|
async def update(self, monitor_id: UUID, data: MonitorUpdate) -> Monitor:
|
|
async with self._lock:
|
|
item = self._items.get(monitor_id)
|
|
if item is None:
|
|
raise MonitorNotFound(monitor_id)
|
|
updated = item.model_copy(update={**data.model_dump(), "updated_at": utc_now()})
|
|
self._items[monitor_id] = updated
|
|
return updated.model_copy(deep=True)
|
|
|
|
async def set_status(self, monitor_id: UUID, status: CurrentStatus) -> CurrentStatus:
|
|
async with self._lock:
|
|
item = self._items.get(monitor_id)
|
|
if item is None:
|
|
raise MonitorNotFound(monitor_id)
|
|
self._items[monitor_id] = item.model_copy(
|
|
update={"current_status": status.model_copy(deep=True), "updated_at": utc_now()}
|
|
)
|
|
return status.model_copy(deep=True)
|
|
|
|
async def delete(self, monitor_id: UUID) -> None:
|
|
async with self._lock:
|
|
if self._items.pop(monitor_id, None) is None:
|
|
raise MonitorNotFound(monitor_id)
|