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

This commit is contained in:
2026-08-09 15:46:42 +00:00
parent 8b43eeda0a
commit d31ef63b2a

View File

@@ -2,69 +2,40 @@ from datetime import UTC, datetime
from enum import StrEnum from enum import StrEnum
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field
def now_utc() -> datetime:
return datetime.now(UTC)
class State(StrEnum): class State(StrEnum):
NEVER_CHECKED = "never_checked" UNKNOWN = "unknown"
UP = "up" UP = "up"
DOWN = "down" DOWN = "down"
ERROR = "error" ERROR = "error"
BLOCKED = "blocked"
class MonitorCreate(BaseModel): class MonitorCreate(BaseModel):
name: str = Field(min_length=1, max_length=100) name: str = Field(min_length=1, max_length=100)
url: HttpUrl url: AnyHttpUrl
expected_status: int = Field(default=200, ge=100, le=599)
class MonitorUpdate(BaseModel): class MonitorUpdate(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=100) name: str | None = Field(default=None, min_length=1, max_length=100)
url: HttpUrl | None = None url: AnyHttpUrl | None = None
expected_status: int | None = Field(default=None, ge=100, le=599)
@model_validator(mode="after")
def not_empty(self) -> "MonitorUpdate":
if not self.model_fields_set:
raise ValueError("at least one field is required")
return self
class CurrentStatus(BaseModel): class CurrentStatus(BaseModel):
model_config = ConfigDict(frozen=True) state: State = State.UNKNOWN
state: State = State.NEVER_CHECKED
checked_at: datetime | None = None checked_at: datetime | None = None
observed_status: int | None = None status_code: int | None = None
latency_ms: float | None = None latency_ms: float | None = Field(default=None, ge=0)
error: str | None = None error: str | None = None
class Monitor(BaseModel): class Monitor(BaseModel):
model_config = ConfigDict(frozen=True) model_config = ConfigDict(from_attributes=True)
id: UUID = Field(default_factory=uuid4) id: UUID = Field(default_factory=uuid4)
name: str name: str
url: HttpUrl url: AnyHttpUrl
expected_status: int created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
created_at: datetime = Field(default_factory=now_utc) updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
updated_at: datetime = Field(default_factory=now_utc)
current_status: CurrentStatus = Field(default_factory=CurrentStatus) current_status: CurrentStatus = Field(default_factory=CurrentStatus)
class CheckResult(CurrentStatus):
monitor_id: UUID
class ErrorBody(BaseModel):
code: str
message: str
class ErrorResponse(BaseModel):
error: ErrorBody