82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
import asyncio
|
|
from uuid import UUID, uuid4
|
|
|
|
from app.models import CurrentStatus, Monitor, MonitorInput, State, utcnow
|
|
|
|
|
|
class NotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
class CapacityError(Exception):
|
|
pass
|
|
|
|
|
|
class StaleCheckError(Exception):
|
|
pass
|
|
|
|
|
|
class MonitorStore:
|
|
def __init__(self, capacity: int) -> None:
|
|
self._items: dict[UUID, Monitor] = {}
|
|
self._lock = asyncio.Lock()
|
|
self._capacity = capacity
|
|
|
|
async def create(self, data: MonitorInput) -> Monitor:
|
|
async with self._lock:
|
|
if len(self._items) >= self._capacity:
|
|
raise CapacityError
|
|
now = utcnow()
|
|
item = Monitor(
|
|
id=uuid4(), name=data.name, url=data.url, created_at=now,
|
|
updated_at=now, current_status=CurrentStatus(),
|
|
)
|
|
self._items[item.id] = item
|
|
return item.model_copy(deep=True)
|
|
|
|
async def list(self) -> list[Monitor]:
|
|
async with self._lock:
|
|
return [item.model_copy(deep=True) for item in self._items.values()]
|
|
|
|
async def get(self, monitor_id: UUID) -> Monitor:
|
|
async with self._lock:
|
|
item = self._items.get(monitor_id)
|
|
if item is None:
|
|
raise NotFoundError
|
|
return item.model_copy(deep=True)
|
|
|
|
async def replace(self, monitor_id: UUID, data: MonitorInput) -> Monitor:
|
|
async with self._lock:
|
|
current = self._items.get(monitor_id)
|
|
if current is None:
|
|
raise NotFoundError
|
|
status = current.current_status
|
|
if str(current.url) != str(data.url):
|
|
status = CurrentStatus(state=State.UNKNOWN)
|
|
item = current.model_copy(update={
|
|
"name": data.name, "url": data.url, "updated_at": utcnow(),
|
|
"current_status": status,
|
|
})
|
|
self._items[monitor_id] = item
|
|
return item.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 NotFoundError
|
|
|
|
async def publish_status(
|
|
self, monitor_id: UUID, expected_url: str, status: CurrentStatus
|
|
) -> Monitor:
|
|
async with self._lock:
|
|
current = self._items.get(monitor_id)
|
|
if current is None:
|
|
raise NotFoundError
|
|
if str(current.url) != expected_url:
|
|
raise StaleCheckError
|
|
item = current.model_copy(update={
|
|
"current_status": status, "updated_at": utcnow()
|
|
})
|
|
self._items[monitor_id] = item
|
|
return item.model_copy(deep=True)
|