126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
from uuid import UUID
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, Response, status
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from app.checker import EndpointChecker
|
|
from app.config import Settings, get_settings
|
|
from app.logging import configure_logging
|
|
from app.models import CheckResult, ErrorResponse, Monitor, MonitorCreate, MonitorUpdate
|
|
from app.store import MonitorNotFoundError, MonitorStore
|
|
|
|
ERROR_RESPONSES = {
|
|
404: {"model": ErrorResponse, "description": "Monitor not found"},
|
|
422: {"model": ErrorResponse, "description": "Invalid request"},
|
|
}
|
|
|
|
|
|
def create_app(
|
|
settings: Settings | None = None,
|
|
store: MonitorStore | None = None,
|
|
checker: EndpointChecker | None = None,
|
|
) -> FastAPI:
|
|
settings = settings or get_settings()
|
|
configure_logging(settings.log_level)
|
|
app = FastAPI(title="Endpoint Monitor Service", version="0.1.0")
|
|
app.state.store = store or MonitorStore()
|
|
app.state.checker = checker or EndpointChecker(
|
|
settings.check_timeout_seconds, settings.max_redirects
|
|
)
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def http_error(_: Request, exc: HTTPException) -> JSONResponse:
|
|
code = "not_found" if exc.status_code == 404 else "http_error"
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={"error": {"code": code, "message": str(exc.detail)}},
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_error(_: Request, __: RequestValidationError) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content={"error": {"code": "invalid_request", "message": "request validation failed"}},
|
|
)
|
|
|
|
def missing() -> HTTPException:
|
|
return HTTPException(status_code=404, detail="monitor not found")
|
|
|
|
@app.get("/health/live", tags=["operations"])
|
|
async def live() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/health/ready", tags=["operations"])
|
|
async def ready() -> dict[str, str]:
|
|
return {"status": "ready", "storage": "process-local-memory"}
|
|
|
|
@app.post(
|
|
"/monitors",
|
|
response_model=Monitor,
|
|
status_code=status.HTTP_201_CREATED,
|
|
responses={422: ERROR_RESPONSES[422]},
|
|
)
|
|
async def create_monitor(data: MonitorCreate) -> Monitor:
|
|
return await app.state.store.create(data)
|
|
|
|
@app.get("/monitors", response_model=list[Monitor])
|
|
async def list_monitors() -> list[Monitor]:
|
|
return await app.state.store.list()
|
|
|
|
@app.get("/monitors/{monitor_id}", response_model=Monitor, responses=ERROR_RESPONSES)
|
|
async def get_monitor(monitor_id: UUID) -> Monitor:
|
|
try:
|
|
return await app.state.store.get(monitor_id)
|
|
except MonitorNotFoundError as exc:
|
|
raise missing() from exc
|
|
|
|
@app.patch("/monitors/{monitor_id}", response_model=Monitor, responses=ERROR_RESPONSES)
|
|
async def update_monitor(monitor_id: UUID, data: MonitorUpdate) -> Monitor:
|
|
try:
|
|
return await app.state.store.update(monitor_id, data)
|
|
except MonitorNotFoundError as exc:
|
|
raise missing() from exc
|
|
|
|
@app.delete(
|
|
"/monitors/{monitor_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
responses={404: ERROR_RESPONSES[404]},
|
|
)
|
|
async def delete_monitor(monitor_id: UUID) -> Response:
|
|
try:
|
|
await app.state.store.delete(monitor_id)
|
|
except MonitorNotFoundError as exc:
|
|
raise missing() from exc
|
|
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
|
|
|
@app.get(
|
|
"/monitors/{monitor_id}/status",
|
|
response_model=CheckResult,
|
|
responses=ERROR_RESPONSES,
|
|
)
|
|
async def current_status(monitor_id: UUID) -> CheckResult:
|
|
try:
|
|
monitor = await app.state.store.get(monitor_id)
|
|
except MonitorNotFoundError as exc:
|
|
raise missing() from exc
|
|
return CheckResult(monitor_id=monitor.id, **monitor.current_status.model_dump())
|
|
|
|
@app.post(
|
|
"/monitors/{monitor_id}/check",
|
|
response_model=CheckResult,
|
|
responses=ERROR_RESPONSES,
|
|
)
|
|
async def check_monitor(monitor_id: UUID) -> CheckResult:
|
|
try:
|
|
monitor = await app.state.store.get(monitor_id)
|
|
result = await app.state.checker.check(monitor)
|
|
await app.state.store.set_status(
|
|
monitor_id, result.model_copy(update={}, deep=True)
|
|
)
|
|
return result
|
|
except MonitorNotFoundError as exc:
|
|
raise missing() from exc
|
|
|
|
return app
|