admin user
56aa5f78a4
Some checks failed
__APP_NAME__ — build & deploy / ¿Credenciales de build listas? (push) Has been cancelled
__APP_NAME__ — build & deploy / Calidad + build + push (push) Has been cancelled
__APP_NAME__ — build & deploy / ¿Cluster del tenant listo? (push) Has been cancelled
__APP_NAME__ — build & deploy / Helm upgrade --install (push) Has been cancelled
129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
"""
|
|
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()
|