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:41:10 +00:00
parent 8c41d3712b
commit 2ec7897acd

View File

@@ -1,9 +1,12 @@
from datetime import datetime, timezone
from enum import StrEnum
from typing import Self
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator
from pydantic import BaseModel, ConfigDict, Field, HttpUrl
def utcnow() -> datetime:
return datetime.now(timezone.utc)
class State(StrEnum):
@@ -13,49 +16,22 @@ class State(StrEnum):
ERROR = "error"
class MonitorInput(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True)
name: str = Field(min_length=1, max_length=100)
url: HttpUrl
class CurrentStatus(BaseModel):
model_config = ConfigDict(extra="forbid")
state: State = State.UNKNOWN
checked_at: datetime | None = None
latency_ms: float | None = Field(default=None, ge=0)
http_status: int | None = Field(default=None, ge=100, le=599)
final_url: str | None = None
error: str | None = None
class MonitorCreate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str = Field(min_length=1, max_length=120)
url: HttpUrl
@model_validator(mode="after")
def safe_url_shape(self) -> Self:
if self.url.scheme not in {"http", "https"}:
raise ValueError("only http and https URLs are supported")
if self.url.username is not None or self.url.password is not None:
raise ValueError("URL credentials are not allowed")
return self
class MonitorUpdate(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str | None = Field(default=None, min_length=1, max_length=120)
url: HttpUrl | None = None
@model_validator(mode="after")
def validate_update(self) -> Self:
if self.name is None and self.url is None:
raise ValueError("at least one field is required")
if self.url is not None:
if self.url.scheme not in {"http", "https"}:
raise ValueError("only http and https URLs are supported")
if self.url.username is not None or self.url.password is not None:
raise ValueError("URL credentials are not allowed")
return self
status_code: int | None = Field(default=None, ge=100, le=599)
error_code: str | None = None
class Monitor(BaseModel):
model_config = ConfigDict(extra="forbid")
id: UUID
name: str
url: HttpUrl
@@ -64,5 +40,14 @@ class Monitor(BaseModel):
current_status: CurrentStatus
def utcnow() -> datetime:
return datetime.now(timezone.utc)
class CheckResult(CurrentStatus):
monitor_id: UUID
class ErrorBody(BaseModel):
code: str
message: str
class ErrorEnvelope(BaseModel):
error: ErrorBody