import logging from contextlib import asynccontextmanager from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from starlette.responses import HTMLResponse, Response from onguard24.config import get_settings from onguard24.db import create_pool, migrate from onguard24.ingress import grafana as grafana_ingress from onguard24.modules import contacts, schedules, statusboard from onguard24.root_html import render_root_page from onguard24.status_snapshot import build as build_status from onguard24 import __version__ as app_version logging.basicConfig(level=logging.INFO) logging.getLogger("httpx").setLevel(logging.WARNING) log = logging.getLogger("onguard24") def parse_addr(http_addr: str) -> tuple[str, int]: s = http_addr.strip() if s.startswith(":"): return "0.0.0.0", int(s[1:]) if ":" in s: h, p = s.rsplit(":", 1) return (h or "0.0.0.0"), int(p) return "0.0.0.0", int(s) @asynccontextmanager async def lifespan(app: FastAPI): settings = get_settings() pool = await create_pool(settings) if pool: await migrate(pool) app.state.pool = pool app.state.settings = settings log.info("onGuard24 started, db=%s", "ok" if pool else "disabled") yield if pool: await pool.close() log.info("database pool closed") def create_app() -> FastAPI: app = FastAPI(title="onGuard24", version=app_version, lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"], allow_credentials=True, allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], ) @app.get("/") async def root(request: Request): if request.query_params.get("format") == "json": data = await build_status(request) data["links"] = { "docs": "/docs", "openapi": "/openapi.json", "health": "/health", "status_api": "/api/v1/status", } return data return HTMLResponse(await render_root_page(request)) @app.get("/favicon.ico", include_in_schema=False) async def favicon(): return Response(status_code=204) @app.get("/health") @app.get("/api/v1/health") async def health(): return {"status": "ok", "service": "onGuard24"} @app.get("/api/v1/status") async def status(request: Request): return await build_status(request) app.include_router(grafana_ingress.router, prefix="/api/v1") app.include_router(schedules.router, prefix="/api/v1/modules/schedules") app.include_router(contacts.router, prefix="/api/v1/modules/contacts") app.include_router(statusboard.router, prefix="/api/v1/modules/statusboard") return app app = create_app() def run() -> None: import uvicorn settings = get_settings() host, port = parse_addr(settings.http_addr) lvl = settings.log_level.upper() uvicorn.run( "onguard24.main:app", host=host, port=port, log_level=lvl.lower() if lvl in {"DEBUG", "INFO", "WARNING", "ERROR"} else "info", ) if __name__ == "__main__": run()