import asyncio from dataclasses import dataclass from uuid import UUID, uuid4 from .models import CheckResult, Monitor, MonitorCreate, MonitorStatus, MonitorUpdate, utcnow @dataclass class MonitorStore: def __post_init__(self) -> None: self._items: dict[UUID, Monitor] = {} self._lock = asyncio.Lock() async def create(self, data: MonitorCreate) -> Monitor: now = utcnow() item = Monitor(id=uuid4(), name=data.name, url=data.url, created_at=now, updated_at=now) async with self._lock: self._items[item.id] = item return item async def list(self) -> list[Monitor]: async with self._lock: return sorted(self._items.values(), key=lambda item: (item.created_at, str(item.id))) async def get(self, monitor_id: UUID) -> Monitor | None: async with self._lock: return self._items.get(monitor_id) async def update(self, monitor_id: UUID, data: MonitorUpdate) -> Monitor | None: async with self._lock: current = self._items.get(monitor_id) if current is None: return None updated = current.model_copy(update={ "name": data.name, "url": data.url, "updated_at": utcnow(), "status": MonitorStatus.unknown, "checked_at": None, "latency_ms": None, "http_status": None, "error": None, }) self._items[monitor_id] = updated return updated async def delete(self, monitor_id: UUID) -> bool: async with self._lock: return self._items.pop(monitor_id, None) is not None async def record(self, monitor_id: UUID, result: CheckResult) -> Monitor | None: async with self._lock: current = self._items.get(monitor_id) if current is None: return None updated = current.model_copy(update={**result.model_dump(), "updated_at": utcnow()}) self._items[monitor_id] = updated return updated