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:15 +00:00
parent e8c5966411
commit adfa4602d5

View File

@@ -1,50 +1,156 @@
import logging import logging
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import AsyncIterator from typing import AsyncIterator
from uuid import UUID
from fastapi import FastAPI, HTTPException, Request from fastapi import FastAPI, Request, Response, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse from fastapi.responses import JSONResponse
from app.api import router from app.checker import (
from app.checker import EndpointChecker CheckNetworkError,
from app.settings import Settings, get_settings CheckTimeoutError,
from app.store import MonitorStore EndpointChecker,
TooManyRedirectsError,
UnsafeTargetError,
)
from app.config import Settings, get_settings
from app.logging_config import configure_logging
from app.models import CheckResult, CurrentStatus, Monitor, MonitorInput, State, utcnow
from app.store import CapacityError, MonitorStore, NotFoundError, StaleCheckError
logger = logging.getLogger("monitor.check")
def configure_logging(level: str) -> None: def error(status_code: int, code: str, message: str) -> JSONResponse:
logging.basicConfig(level=level.upper(), format="%(message)s") return JSONResponse(status_code=status_code, content={
"error": {"code": code, "message": message}
})
def create_app(settings: Settings | None = None) -> FastAPI: def create_app(settings: Settings | None = None) -> FastAPI:
config = settings or get_settings() config = settings or get_settings()
@asynccontextmanager @asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]: async def lifespan(app: FastAPI) -> AsyncIterator[None]:
configure_logging(config.log_level) configure_logging(config.log_level)
application.state.store = MonitorStore() app.state.store = MonitorStore(config.max_monitors)
application.state.checker = EndpointChecker(config) app.state.checker = EndpointChecker(
config.request_timeout_seconds, config.max_redirects
)
app.state.ready = True
yield yield
app.state.ready = False
application = FastAPI(title="Endpoint Monitor", version="1.0.0", lifespan=lifespan) api = FastAPI(title="Endpoint Monitor API", version="1.0.0", lifespan=lifespan)
application.include_router(router)
@application.exception_handler(HTTPException) @api.exception_handler(NotFoundError)
async def http_error(_request: Request, exc: HTTPException) -> JSONResponse: async def not_found(_: Request, __: NotFoundError) -> JSONResponse:
if isinstance(exc.detail, tuple): return error(404, "monitor_not_found", "monitor does not exist")
code, message = exc.detail
else:
code, message = "http_error", str(exc.detail)
return JSONResponse(status_code=exc.status_code,
content={"error": {"code": code, "message": message}},
headers=exc.headers)
@application.exception_handler(RequestValidationError) @api.exception_handler(CapacityError)
async def validation_error(_request: Request, _exc: RequestValidationError) -> JSONResponse: async def capacity(_: Request, __: CapacityError) -> JSONResponse:
return JSONResponse(status_code=422, content={"error": { return error(409, "capacity_exceeded", "monitor capacity reached")
"code": "validation_error", "message": "Request validation failed"}})
return application @api.get("/healthz")
async def health() -> dict[str, str]:
return {"status": "ok"}
@api.get("/readyz")
async def readiness(request: Request) -> JSONResponse | dict[str, str]:
if not getattr(request.app.state, "ready", False):
return error(503, "not_ready", "service is not ready")
return {"status": "ok"}
@api.post("/monitors", response_model=Monitor, status_code=status.HTTP_201_CREATED)
async def create(data: MonitorInput, request: Request) -> Monitor:
return await request.app.state.store.create(data)
@api.get("/monitors", response_model=list[Monitor])
async def list_monitors(request: Request) -> list[Monitor]:
return await request.app.state.store.list()
@api.get("/monitors/{monitor_id}", response_model=Monitor)
async def get_monitor(monitor_id: UUID, request: Request) -> Monitor:
return await request.app.state.store.get(monitor_id)
@api.put("/monitors/{monitor_id}", response_model=Monitor)
async def replace(monitor_id: UUID, data: MonitorInput, request: Request) -> Monitor:
return await request.app.state.store.replace(monitor_id, data)
@api.delete("/monitors/{monitor_id}", status_code=204)
async def delete(monitor_id: UUID, request: Request) -> Response:
await request.app.state.store.delete(monitor_id)
return Response(status_code=204)
@api.get("/monitors/{monitor_id}/status", response_model=CurrentStatus)
async def current_status(monitor_id: UUID, request: Request) -> CurrentStatus:
return (await request.app.state.store.get(monitor_id)).current_status
@api.post("/monitors/{monitor_id}/check", response_model=CheckResult)
async def check(monitor_id: UUID, request: Request) -> CheckResult | JSONResponse:
store: MonitorStore = request.app.state.store
item = await store.get(monitor_id)
checked_url = str(item.url)
logger.info("check_started", extra={"fields": {"url": checked_url, "monitor_id": monitor_id}})
try:
outcome = await request.app.state.checker.check(checked_url)
current = CurrentStatus(
state=State.UP if 200 <= outcome.status_code < 400 else State.DOWN,
checked_at=utcnow(), latency_ms=outcome.latency_ms,
status_code=outcome.status_code,
)
await store.publish_status(monitor_id, checked_url, current)
logger.info("check_finished", extra={"fields": {
"url": checked_url, "monitor_id": monitor_id,
"status_code": outcome.status_code, "latency_ms": outcome.latency_ms,
}})
return CheckResult(monitor_id=monitor_id, **current.model_dump())
except UnsafeTargetError:
return await publish_error(store, monitor_id, checked_url, "unsafe_target", 400)
except CheckTimeoutError as exc:
return await publish_error(
store, monitor_id, checked_url, "check_timeout", 504, exc.latency_ms
)
except CheckNetworkError as exc:
return await publish_error(
store, monitor_id, checked_url, "network_error", 502, exc.latency_ms
)
except TooManyRedirectsError as exc:
return await publish_error(
store, monitor_id, checked_url, "redirect_limit", 502, exc.latency_ms
)
except StaleCheckError:
return error(409, "stale_check", "monitor URL changed during check")
return api
async def publish_error(
store: MonitorStore,
monitor_id: UUID,
checked_url: str,
code: str,
http_status: int,
latency_ms: float | None = None,
) -> JSONResponse:
current = CurrentStatus(
state=State.ERROR, checked_at=utcnow(), latency_ms=latency_ms, error_code=code
)
try:
await store.publish_status(monitor_id, checked_url, current)
except StaleCheckError:
return error(409, "stale_check", "monitor URL changed during check")
logger.warning("check_failed", extra={"fields": {
"url": checked_url, "monitor_id": monitor_id, "error_code": code,
"latency_ms": latency_ms,
}})
messages = {
"unsafe_target": "target is not permitted",
"check_timeout": "target request timed out",
"network_error": "target request failed",
"redirect_limit": "target exceeded redirect limit",
}
return error(http_status, code, messages[code])
app = create_app() app = create_app()