egutierrez
be5a7b582e
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).
88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
"""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)
|