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
Some checks failed
ci / validate (push) Has been cancelled
This commit is contained in:
155
app/main.py
155
app/main.py
@@ -1,156 +1,3 @@
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import AsyncIterator
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import FastAPI, Request, Response, status
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
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 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(app: FastAPI) -> AsyncIterator[None]:
|
||||
configure_logging(config.log_level)
|
||||
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
|
||||
|
||||
api = FastAPI(title="Endpoint Monitor API", version="1.0.0", lifespan=lifespan)
|
||||
|
||||
@api.exception_handler(NotFoundError)
|
||||
async def not_found(_: Request, __: NotFoundError) -> JSONResponse:
|
||||
return error(404, "monitor_not_found", "monitor does not exist")
|
||||
|
||||
@api.exception_handler(CapacityError)
|
||||
async def capacity(_: Request, __: CapacityError) -> JSONResponse:
|
||||
return error(409, "capacity_exceeded", "monitor capacity reached")
|
||||
|
||||
@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])
|
||||
|
||||
from app.api import create_app
|
||||
|
||||
app = create_app()
|
||||
|
||||
Reference in New Issue
Block a user