""" Aplicación FastAPI — entry point. Endpoints obligatorios del shape web-backend/python (ver coralware-shape.yaml): - / → página de bienvenida (instrucciones de uso del tenant; tu código la reemplaza) - /health → liveness + readiness - /docs → Swagger UI (auth opcional) - /demo → frontend estático servido desde STATIC_DEMO_PATH - /metrics → Prometheus Endpoints de dominio: agregar en app/routers/ y registrar en _register_routers(). La ruta raíz `/` sirve la página de bienvenida por defecto; en cuanto agregues tu propio handler de `/` (o lo registres en app/routers/) tu desarrollo la reemplaza. """ import logging import os from fastapi import Depends, FastAPI, HTTPException, Request, status from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from prometheus_client import CONTENT_TYPE_LATEST, generate_latest from app.db import db_alive from app.settings import settings from app.welcome import WELCOME_HTML logging.basicConfig( level=getattr(logging, settings.LOG_LEVEL, logging.INFO), format="%(asctime)s [%(levelname)s] [cid=%(correlation_id)s] %(message)s", datefmt="%Y-%m-%dT%H:%M:%SZ", ) class CorrelationIdFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: if not hasattr(record, "correlation_id"): record.correlation_id = "-" return True logging.getLogger().addFilter(CorrelationIdFilter()) log = logging.getLogger("__APP_NAME__") def _require_auth_dep() -> None: """ Dependency placeholder para /docs cuando OPENAPI_REQUIRE_AUTH=true. Implementación real: validar JWT contra KEYCLOAK_ISSUER (ADR-029 R-004). """ if not settings.OPENAPI_REQUIRE_AUTH: return raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Auth requerido — implementar validación JWT contra Keycloak", ) def create_app() -> FastAPI: app = FastAPI( title="__APP_NAME__", version="0.1.0", dependencies=([Depends(_require_auth_dep)] if settings.OPENAPI_REQUIRE_AUTH else []), ) @app.middleware("http") async def correlation_id_middleware(request: Request, call_next): cid = request.headers.get(settings.CORRELATION_ID_HEADER) or _generate_cid() request.state.correlation_id = cid response = await call_next(request) response.headers[settings.CORRELATION_ID_HEADER] = cid return response @app.get("/", response_class=HTMLResponse, include_in_schema=False) async def welcome() -> HTMLResponse: """ Página de bienvenida por defecto del Golden Path. El tenant la ve en cuanto su plataforma queda lista. En cuanto agrega su propio handler de `/` (su desarrollo), esta página es reemplazada. """ return HTMLResponse(content=WELCOME_HTML) @app.get("/health", tags=["platform"]) async def health() -> JSONResponse: db_status = "up" if db_alive() else "down" code = 200 if db_status == "up" else 503 return JSONResponse( status_code=code, content={"status": "ok" if code == 200 else "degraded", "db": db_status}, ) @app.get("/metrics", tags=["platform"]) async def metrics() -> Response: return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST) if os.path.isdir(settings.STATIC_DEMO_PATH): app.mount( "/demo", StaticFiles(directory=settings.STATIC_DEMO_PATH, html=True), name="demo", ) else: log.warning( "STATIC_DEMO_PATH=%s no existe — endpoint /demo no será servido", settings.STATIC_DEMO_PATH, ) _register_routers(app) return app def _generate_cid() -> str: """Genera correlation_id si el cliente no envió uno.""" import time import uuid return f"__APP_NAME__-{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}" def _register_routers(app: FastAPI) -> None: """ Registra los routers de dominio. Agrega aquí los include_router(...) conforme tu app crezca. """ pass app = create_app()