egutierrez
6a05abff03
Tres pipelines Python para gestionar operations.db en Metabase: - metabase_add_ops_db: registra la operations.db de una app como database SQLite - metabase_create_ops_dashboard: genera dashboard operativo con 14 cards (KPIs, distribución, executions, assertions) para cualquier app - metabase_fix_permissions: arregla SQLITE_READONLY_DIRECTORY haciendo chmod 777/666 sin chown (que se propaga al host via bind mount) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
"""Pipeline: registra la operations.db de una app en Metabase.
|
|
|
|
Compone: metabase_auth + metabase_list_databases + metabase_add_database.
|
|
|
|
Requiere que el contenedor Metabase tenga montado el directorio de la app
|
|
como volumen RW para que SQLite pueda crear journal files.
|
|
|
|
Uso:
|
|
python metabase_add_ops_db.py <app_name>
|
|
python metabase_add_ops_db.py docker_tui
|
|
python metabase_add_ops_db.py --list
|
|
"""
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
|
|
from metabase.client import metabase_auth
|
|
from metabase import metabase_list_databases, metabase_add_database
|
|
|
|
METABASE_URL = os.environ.get("METABASE_URL", "http://localhost:3000")
|
|
EMAIL = os.environ.get("METABASE_ADMIN_EMAIL", "admin@fnregistry.local")
|
|
PASSWORD = os.environ.get("METABASE_ADMIN_PASSWORD", "FnRegistry2024!")
|
|
|
|
CONTAINER_DATA_PREFIX = "/data/ops-"
|
|
|
|
|
|
def find_apps(registry_root: str) -> list[str]:
|
|
"""Lista apps que tienen operations.db."""
|
|
apps_dir = os.path.join(registry_root, "apps")
|
|
if not os.path.isdir(apps_dir):
|
|
return []
|
|
return sorted(
|
|
d
|
|
for d in os.listdir(apps_dir)
|
|
if os.path.isfile(os.path.join(apps_dir, d, "operations.db"))
|
|
)
|
|
|
|
|
|
def db_name_for_app(app_name: str) -> str:
|
|
return f"ops-{app_name.replace('_', '-')}"
|
|
|
|
|
|
def container_path_for_app(app_name: str) -> str:
|
|
return f"{CONTAINER_DATA_PREFIX}{app_name.replace('_', '-')}/operations.db"
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Registra operations.db de una app en Metabase")
|
|
parser.add_argument("app_name", nargs="?", help="Nombre de la app (directorio en apps/)")
|
|
parser.add_argument("--list", action="store_true", help="Lista apps disponibles y su estado en Metabase")
|
|
args = parser.parse_args()
|
|
|
|
registry_root = os.environ.get(
|
|
"FN_REGISTRY_ROOT",
|
|
os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")),
|
|
)
|
|
|
|
if args.list:
|
|
client = metabase_auth(METABASE_URL, EMAIL, PASSWORD)
|
|
dbs = metabase_list_databases(client)
|
|
db_names = {d["name"] for d in dbs}
|
|
apps = find_apps(registry_root)
|
|
print(f"Apps con operations.db ({len(apps)}):\n")
|
|
for app in apps:
|
|
name = db_name_for_app(app)
|
|
status = "registrada" if name in db_names else "NO registrada"
|
|
print(f" {app:30s} -> {name:30s} [{status}]")
|
|
client.close()
|
|
return
|
|
|
|
if not args.app_name:
|
|
parser.error("Se requiere app_name o --list")
|
|
|
|
app_name = args.app_name
|
|
ops_path = os.path.join(registry_root, "apps", app_name, "operations.db")
|
|
if not os.path.isfile(ops_path):
|
|
print(f"ERROR: No existe {ops_path}")
|
|
print(f"Apps disponibles: {', '.join(find_apps(registry_root))}")
|
|
sys.exit(1)
|
|
|
|
name = db_name_for_app(app_name)
|
|
container_path = container_path_for_app(app_name)
|
|
|
|
print(f"Registrando {app_name} en Metabase...")
|
|
print(f" Nombre: {name}")
|
|
print(f" Path: {container_path}")
|
|
|
|
client = metabase_auth(METABASE_URL, EMAIL, PASSWORD)
|
|
|
|
# Verificar si ya existe
|
|
dbs = metabase_list_databases(client)
|
|
for db in dbs:
|
|
if db["name"] == name:
|
|
print(f" Ya existe con id={db['id']}, nada que hacer.")
|
|
client.close()
|
|
return
|
|
|
|
# Registrar
|
|
result = metabase_add_database(client, name, "sqlite", {"db": container_path})
|
|
db_id = result["id"]
|
|
print(f" Registrada con id={db_id}")
|
|
|
|
# Recordar: el contenedor necesita el volumen montado
|
|
print(f"\nIMPORTANTE: El contenedor Metabase debe tener montado:")
|
|
print(f" -v {os.path.join(registry_root, 'apps', app_name)}:{CONTAINER_DATA_PREFIX}{app_name.replace('_', '-')}")
|
|
print(f"\nSi el mount no existe, hay que recrear el contenedor.")
|
|
print(f"\nDatabase disponible en: {METABASE_URL}/question#new?database={db_id}")
|
|
|
|
client.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|