59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
import asyncio
|
|
from uuid import UUID, uuid4
|
|
|
|
from .models import CheckSnapshot, Monitor, MonitorInput, now
|
|
|
|
|
|
class StoreFullError(Exception):
|
|
pass
|
|
|
|
|
|
class MonitorStore:
|
|
def __init__(self, max_monitors: int) -> None:
|
|
self._items: dict[UUID, Monitor] = {}
|
|
self._lock = asyncio.Lock()
|
|
self._max = max_monitors
|
|
|
|
async def create(self, data: MonitorInput) -> Monitor:
|
|
async with self._lock:
|
|
if len(self._items) >= self._max:
|
|
raise StoreFullError
|
|
timestamp = now()
|
|
item = Monitor(id=uuid4(), **data.model_dump(), created_at=timestamp,
|
|
updated_at=timestamp, revision=1, status=CheckSnapshot())
|
|
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)
|
|
|
|
async def get(self, monitor_id: UUID) -> Monitor | None:
|
|
async with self._lock:
|
|
return self._items.get(monitor_id)
|
|
|
|
async def replace(self, monitor_id: UUID, data: MonitorInput) -> Monitor | None:
|
|
async with self._lock:
|
|
old = self._items.get(monitor_id)
|
|
if old is None:
|
|
return None
|
|
item = old.model_copy(update={**data.model_dump(), "updated_at": now(),
|
|
"revision": old.revision + 1,
|
|
"status": CheckSnapshot()})
|
|
self._items[monitor_id] = item
|
|
return item
|
|
|
|
async def delete(self, monitor_id: UUID) -> bool:
|
|
async with self._lock:
|
|
return self._items.pop(monitor_id, None) is not None
|
|
|
|
async def set_status(self, monitor_id: UUID, revision: int,
|
|
status: CheckSnapshot) -> Monitor | None:
|
|
async with self._lock:
|
|
old = self._items.get(monitor_id)
|
|
if old is None or old.revision != revision:
|
|
return None
|
|
item = old.model_copy(update={"status": status, "updated_at": now()})
|
|
self._items[monitor_id] = item
|
|
return item
|