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 are pending
ci / test (push) Has started running

This commit is contained in:
2026-08-09 16:00:49 +00:00
parent c79b40fbdc
commit 79bd2939c3

View File

@@ -1,33 +1,34 @@
from datetime import datetime from datetime import datetime
from enum import StrEnum from typing import Literal
from uuid import UUID from uuid import UUID
from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, model_validator
StatusState = Literal["unknown", "up", "down", "error"]
class MonitorState(StrEnum): class CheckStatus(BaseModel):
UNKNOWN = "unknown" state: StatusState = "unknown"
UP = "up" checked_at: datetime | None = None
DOWN = "down" status_code: int | None = None
ERROR = "error" latency_ms: float | None = Field(default=None, ge=0)
BLOCKED = "blocked" error: str | None = None
class MonitorCreate(BaseModel): class MonitorCreate(BaseModel):
name: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=200)
url: AnyHttpUrl url: AnyHttpUrl
class MonitorUpdate(MonitorCreate): class MonitorUpdate(BaseModel):
pass name: str | None = Field(default=None, min_length=1, max_length=200)
url: AnyHttpUrl | None = None
@model_validator(mode="after")
class StatusSnapshot(BaseModel): def not_empty(self) -> "MonitorUpdate":
state: MonitorState = MonitorState.UNKNOWN if self.name is None and self.url is None:
checked_at: datetime | None = None raise ValueError("at least one field is required")
latency_ms: float | None = Field(default=None, ge=0) return self
http_status: int | None = Field(default=None, ge=100, le=599)
error: str | None = None
class Monitor(BaseModel): class Monitor(BaseModel):
@@ -38,8 +39,13 @@ class Monitor(BaseModel):
url: AnyHttpUrl url: AnyHttpUrl
created_at: datetime created_at: datetime
updated_at: datetime updated_at: datetime
status: StatusSnapshot status: CheckStatus
class Health(BaseModel): class CheckResult(CheckStatus):
status: str monitor_id: UUID
status_applied: bool = True
class HealthResponse(BaseModel):
status: Literal["ok", "ready"]