decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project.
Some checks failed
ci / validate (push) Has been cancelled
Some checks failed
ci / validate (push) Has been cancelled
This commit is contained in:
94
app/store.py
94
app/store.py
@@ -1,81 +1,65 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
from uuid import UUID, uuid4
|
from uuid import UUID
|
||||||
|
|
||||||
from app.models import CurrentStatus, Monitor, MonitorInput, State, utcnow
|
from app.models import CurrentStatus, Monitor, MonitorCreate, MonitorUpdate, now_utc
|
||||||
|
|
||||||
|
|
||||||
class NotFoundError(Exception):
|
class MonitorNotFoundError(KeyError):
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class CapacityError(Exception):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class StaleCheckError(Exception):
|
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class MonitorStore:
|
class MonitorStore:
|
||||||
def __init__(self, capacity: int) -> None:
|
"""Concurrency-safe, process-local monitor storage.
|
||||||
|
|
||||||
|
Models are frozen and every read returns an immutable value, so callers cannot mutate
|
||||||
|
state outside the lock. All read-modify-write operations happen under one asyncio lock.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
self._items: dict[UUID, Monitor] = {}
|
self._items: dict[UUID, Monitor] = {}
|
||||||
self._lock = asyncio.Lock()
|
self._lock = asyncio.Lock()
|
||||||
self._capacity = capacity
|
|
||||||
|
|
||||||
async def create(self, data: MonitorInput) -> Monitor:
|
async def create(self, data: MonitorCreate) -> Monitor:
|
||||||
|
monitor = Monitor(**data.model_dump())
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
if len(self._items) >= self._capacity:
|
self._items[monitor.id] = monitor
|
||||||
raise CapacityError
|
return monitor
|
||||||
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 def list(self) -> list[Monitor]:
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
return [item.model_copy(deep=True) for item in self._items.values()]
|
return list(self._items.values())
|
||||||
|
|
||||||
async def get(self, monitor_id: UUID) -> Monitor:
|
async def get(self, monitor_id: UUID) -> Monitor:
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
item = self._items.get(monitor_id)
|
try:
|
||||||
if item is None:
|
return self._items[monitor_id]
|
||||||
raise NotFoundError
|
except KeyError as exc:
|
||||||
return item.model_copy(deep=True)
|
raise MonitorNotFoundError(monitor_id) from exc
|
||||||
|
|
||||||
async def replace(self, monitor_id: UUID, data: MonitorInput) -> Monitor:
|
async def update(self, monitor_id: UUID, data: MonitorUpdate) -> Monitor:
|
||||||
|
changes = data.model_dump(exclude_unset=True, exclude_none=True)
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
current = self._items.get(monitor_id)
|
try:
|
||||||
if current is None:
|
existing = self._items[monitor_id]
|
||||||
raise NotFoundError
|
except KeyError as exc:
|
||||||
status = current.current_status
|
raise MonitorNotFoundError(monitor_id) from exc
|
||||||
if str(current.url) != str(data.url):
|
updated = existing.model_copy(update={**changes, "updated_at": now_utc()})
|
||||||
status = CurrentStatus(state=State.UNKNOWN)
|
self._items[monitor_id] = updated
|
||||||
item = current.model_copy(update={
|
return updated
|
||||||
"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 def delete(self, monitor_id: UUID) -> None:
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
if self._items.pop(monitor_id, None) is None:
|
if self._items.pop(monitor_id, None) is None:
|
||||||
raise NotFoundError
|
raise MonitorNotFoundError(monitor_id)
|
||||||
|
|
||||||
async def publish_status(
|
async def set_status(self, monitor_id: UUID, status: CurrentStatus) -> Monitor:
|
||||||
self, monitor_id: UUID, expected_url: str, status: CurrentStatus
|
|
||||||
) -> Monitor:
|
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
current = self._items.get(monitor_id)
|
try:
|
||||||
if current is None:
|
existing = self._items[monitor_id]
|
||||||
raise NotFoundError
|
except KeyError as exc:
|
||||||
if str(current.url) != expected_url:
|
raise MonitorNotFoundError(monitor_id) from exc
|
||||||
raise StaleCheckError
|
updated = existing.model_copy(
|
||||||
item = current.model_copy(update={
|
update={"current_status": status, "updated_at": now_utc()}
|
||||||
"current_status": status, "updated_at": utcnow()
|
)
|
||||||
})
|
self._items[monitor_id] = updated
|
||||||
self._items[monitor_id] = item
|
return updated
|
||||||
return item.model_copy(deep=True)
|
|
||||||
|
|||||||
Reference in New Issue
Block a user