feat: sincronizar scaffold completo del shape web-backend/python
Some checks failed
__APP_NAME__ — build & deploy / build (push) Failing after 0s
__APP_NAME__ — build & deploy / helm-deploy (push) Has been skipped

Sincroniza el scaffold canónico desde templates/web-backend-python/ del
repo IDP. Reemplaza el placeholder mínimo previo (.gitea/workflows/ci.yaml,
README.md, .gitignore, catalog-info.yaml) por el scaffold completo
documentado en ADR-053 §3:

- code/ — FastAPI app con /health (db check), /metrics, /demo estático,
  middleware correlation_id, settings Pydantic, db.py con SQLAlchemy
- code/alembic/ — migración inicial vacía + env.py usando MIGRATION_DATABASE_URL
- code/tests/ — pytest + httpx TestClient (test_health, test_demo)
- helm/APP_NAME/ — chart con Deployment, Service, IngressRoute Traefik,
  NetworkPolicy, ServiceMonitor, Job alembic-upgrade como hook
  pre-upgrade,pre-install (backoffLimit: 0)
- coralware-shape.yaml — manifiesto v1alpha1 del contrato del shape
- claim-mariadb.yaml — AddonClaim declarativo (uso futuro ADR-052)
- catalog-info.yaml — entidad Backstage actualizada
- .gitea/workflows/APP_NAME-build.yaml — invoca reusable workflow ADR-038
  + job helm-deploy con KUBECONFIG_TENANT_B64

Placeholders __APP_NAME__, __TIER__, __TENANT_ID__, __TENANT_ID_NODASHES__
serán sustituidos server-side por repo-provisioner cuando materialice el
template para un tenant (pendiente — ADR-053 §11.5).

El openspec/ del golden path NO se sincroniza al repo Gitea (gobernanza
interna de Coralware).

Refs: ADR-052, ADR-053, poc-nexobms.md (PoC NexoBMS Demetrio).
This commit is contained in:
2026-05-06 17:48:13 -05:00
parent 00a801f525
commit fcec92ea1d
35 changed files with 1216 additions and 32 deletions

0
code/app/__init__.py Normal file
View File

40
code/app/db.py Normal file
View File

@@ -0,0 +1,40 @@
"""
Capa de conexión SQLAlchemy.
- engine único por proceso, configurado al startup
- sesiones por request (dependency injection FastAPI)
- ping ligero para /health
"""
from collections.abc import Iterator
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker
from app.settings import settings
engine = create_engine(
settings.DATABASE_URL,
pool_pre_ping=True,
pool_recycle=3600,
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_session() -> Iterator[Session]:
"""Dependency FastAPI: una sesión por request, cierre garantizado."""
db = SessionLocal()
try:
yield db
finally:
db.close()
def db_alive() -> bool:
"""Ping de conexión usado por /health. Retorna False ante cualquier excepción."""
try:
with engine.connect() as conn:
conn.execute(text("SELECT 1"))
return True
except Exception:
return False

115
code/app/main.py Normal file
View File

@@ -0,0 +1,115 @@
"""
Aplicación FastAPI — entry point.
Endpoints obligatorios del shape web-backend/python (ver coralware-shape.yaml):
- /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().
"""
import logging
import os
from fastapi import Depends, FastAPI, HTTPException, Request, status
from fastapi.responses import 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
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("/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()

View File

@@ -0,0 +1,8 @@
# Modelos SQLAlchemy. Define tu Base aquí y heredala en cada modelo:
#
# from sqlalchemy.orm import DeclarativeBase
# class Base(DeclarativeBase):
# pass
#
# Tras crear modelos, genera la migración con:
# alembic revision --autogenerate -m "descripción del cambio"

View File

@@ -0,0 +1,2 @@
# Routers de dominio. Crea un módulo por agregado (users.py, orders.py, etc.)
# y regístralos en app/main.py:_register_routers().

32
code/app/settings.py Normal file
View File

@@ -0,0 +1,32 @@
"""
Configuración runtime via variables de entorno.
Toda la configuración pasa por aquí — no leas env vars dispersos en otros módulos.
"""
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=None, case_sensitive=True)
DATABASE_URL: str = Field(..., description="Connection string runtime de la app")
MIGRATION_DATABASE_URL: str = Field(
default="",
description="Connection string para Alembic; si vacío, usa DATABASE_URL",
)
LOG_LEVEL: str = "INFO"
CORRELATION_ID_HEADER: str = "X-Correlation-ID"
OPENAPI_REQUIRE_AUTH: bool = False
KEYCLOAK_ISSUER: str = ""
STATIC_DEMO_PATH: str = "/app/static/demo"
@property
def migration_url(self) -> str:
return self.MIGRATION_DATABASE_URL or self.DATABASE_URL
settings = Settings()

View File

@@ -0,0 +1,41 @@
<!DOCTYPE html>
<html lang="es">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>__APP_NAME__ — demo</title>
<style>
body { font-family: system-ui, sans-serif; max-width: 720px; margin: 4rem auto; padding: 0 1rem; }
code { background: #f4f4f4; padding: 0.15rem 0.4rem; border-radius: 3px; }
.status { padding: 0.5rem 1rem; border-radius: 4px; margin-top: 1rem; }
.ok { background: #d4edda; color: #155724; }
.err { background: #f8d7da; color: #721c24; }
</style>
</head>
<body>
<h1>__APP_NAME__</h1>
<p>App generada desde el Golden Path <code>web-backend/python</code>.</p>
<p>
Este es el frontend estático servido en <code>/demo</code>. Modifica
<code>code/app/static/demo/</code> para reemplazarlo con tu UI.
</p>
<div id="health-status" class="status">cargando…</div>
<script>
fetch('/health')
.then(r => r.json())
.then(data => {
const el = document.getElementById('health-status');
const ok = data.status === 'ok';
el.className = 'status ' + (ok ? 'ok' : 'err');
el.textContent = `Estado: ${data.status} · DB: ${data.db}`;
})
.catch(err => {
const el = document.getElementById('health-status');
el.className = 'status err';
el.textContent = 'No se pudo contactar a /health';
});
</script>
</body>
</html>