54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
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
|