feat: funciones Python para API Metabase

Añade módulo Python con funciones para la API de Metabase en dominio infra.
Incluye cliente HTTP, auth, y CRUD de cards, dashboards y users.
Proyecto gestionado con uv (pyproject.toml).
This commit is contained in:
2026-03-28 20:32:28 +01:00
parent 9e6bea681f
commit be5a7b582e
33 changed files with 1325 additions and 0 deletions
+87
View File
@@ -0,0 +1,87 @@
"""Cliente base para la API REST de Metabase."""
import httpx
class MetabaseClient:
"""Cliente HTTP para una instancia Metabase.
Attributes:
base_url: URL base sin trailing slash (ej: "http://localhost:3000").
token: Session token o API key.
_http: Cliente httpx reutilizable con headers de auth.
"""
def __init__(self, base_url: str, token: str) -> None:
self.base_url = base_url.rstrip("/")
self.token = token
self._http = httpx.Client(
base_url=self.base_url,
headers={
"Content-Type": "application/json",
"X-Metabase-Session": token,
},
timeout=30.0,
)
def request(self, method: str, path: str, **kwargs) -> dict | list | None:
"""Ejecuta una peticion HTTP contra la API de Metabase.
Args:
method: HTTP method (GET, POST, PUT, DELETE).
path: Ruta relativa (ej: "/api/user").
**kwargs: Argumentos extra para httpx (json, params, etc.).
Returns:
Respuesta deserializada como dict/list, o None si el body esta vacio.
Raises:
httpx.HTTPStatusError: Si el status code no es 2xx.
"""
resp = self._http.request(method, path, **kwargs)
resp.raise_for_status()
if not resp.content:
return None
return resp.json()
def close(self) -> None:
"""Cierra el cliente HTTP."""
self._http.close()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
def metabase_auth(base_url: str, email: str, password: str) -> MetabaseClient:
"""Autentica contra Metabase con email y password.
Crea una sesion via POST /api/session y retorna un MetabaseClient
con el session token listo para usar. El token expira en 14 dias
por defecto (configurable con MAX_SESSION_AGE en Metabase).
Args:
base_url: URL base de la instancia (ej: "http://localhost:3000").
email: Email del usuario Metabase.
password: Password del usuario.
Returns:
MetabaseClient autenticado con session token.
Raises:
httpx.HTTPStatusError: Si las credenciales son invalidas (401)
o hay rate limiting.
Example:
>>> client = metabase_auth("http://localhost:3000", "admin@example.com", "pass")
>>> # client listo para usar con todas las funciones CRUD
"""
resp = httpx.post(
f"{base_url.rstrip('/')}/api/session",
json={"username": email, "password": password},
)
resp.raise_for_status()
token = resp.json()["id"]
return MetabaseClient(base_url, token)