From adfa4602d59ee3f300330bdf309aaa9551a9605a Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 15:41:15 +0000 Subject: [PATCH] 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. --- app/main.py | 160 +++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 133 insertions(+), 27 deletions(-) diff --git a/app/main.py b/app/main.py index dcfbbbb..0380cc6 100644 --- a/app/main.py +++ b/app/main.py @@ -1,50 +1,156 @@ import logging from contextlib import asynccontextmanager from typing import AsyncIterator +from uuid import UUID -from fastapi import FastAPI, HTTPException, Request -from fastapi.exceptions import RequestValidationError +from fastapi import FastAPI, Request, Response, status from fastapi.responses import JSONResponse -from app.api import router -from app.checker import EndpointChecker -from app.settings import Settings, get_settings -from app.store import MonitorStore +from app.checker import ( + CheckNetworkError, + CheckTimeoutError, + 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: - logging.basicConfig(level=level.upper(), format="%(message)s") +def error(status_code: int, code: str, message: str) -> JSONResponse: + return JSONResponse(status_code=status_code, content={ + "error": {"code": code, "message": message} + }) def create_app(settings: Settings | None = None) -> FastAPI: config = settings or get_settings() @asynccontextmanager - async def lifespan(application: FastAPI) -> AsyncIterator[None]: + async def lifespan(app: FastAPI) -> AsyncIterator[None]: configure_logging(config.log_level) - application.state.store = MonitorStore() - application.state.checker = EndpointChecker(config) + app.state.store = MonitorStore(config.max_monitors) + app.state.checker = EndpointChecker( + config.request_timeout_seconds, config.max_redirects + ) + app.state.ready = True yield + app.state.ready = False - application = FastAPI(title="Endpoint Monitor", version="1.0.0", lifespan=lifespan) - application.include_router(router) + api = FastAPI(title="Endpoint Monitor API", version="1.0.0", lifespan=lifespan) - @application.exception_handler(HTTPException) - async def http_error(_request: Request, exc: HTTPException) -> JSONResponse: - if isinstance(exc.detail, tuple): - 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) + @api.exception_handler(NotFoundError) + async def not_found(_: Request, __: NotFoundError) -> JSONResponse: + return error(404, "monitor_not_found", "monitor does not exist") - @application.exception_handler(RequestValidationError) - async def validation_error(_request: Request, _exc: RequestValidationError) -> JSONResponse: - return JSONResponse(status_code=422, content={"error": { - "code": "validation_error", "message": "Request validation failed"}}) + @api.exception_handler(CapacityError) + async def capacity(_: Request, __: CapacityError) -> JSONResponse: + return error(409, "capacity_exceeded", "monitor capacity reached") - 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()