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).
154 lines
4.7 KiB
Python
154 lines
4.7 KiB
Python
"""CRUD de usuarios de Metabase."""
|
|
|
|
from .client import MetabaseClient
|
|
|
|
|
|
def metabase_list_users(
|
|
client: MetabaseClient,
|
|
status: str = "",
|
|
query: str = "",
|
|
limit: int = 0,
|
|
offset: int = 0,
|
|
) -> dict:
|
|
"""Lista usuarios de Metabase con filtros opcionales.
|
|
|
|
Endpoint: GET /api/user. Requiere permisos de superusuario.
|
|
|
|
Args:
|
|
client: Cliente autenticado.
|
|
status: "active" (default), "deactivated" o "all".
|
|
query: Filtro por nombre o email.
|
|
limit: Tamanio de pagina (0 = default Metabase).
|
|
offset: Offset para paginacion.
|
|
|
|
Returns:
|
|
Dict con estructura paginada:
|
|
- data: lista de usuarios (id, email, first_name, last_name, is_superuser, etc.)
|
|
- total: numero total de usuarios que coinciden
|
|
- limit: tamanio de pagina usado
|
|
- offset: offset usado
|
|
|
|
Example:
|
|
>>> users = metabase_list_users(client, status="active", query="john@")
|
|
>>> for u in users["data"]:
|
|
... print(u["email"], u["first_name"])
|
|
"""
|
|
params = {}
|
|
if status:
|
|
params["status"] = status
|
|
if query:
|
|
params["query"] = query
|
|
if limit > 0:
|
|
params["limit"] = limit
|
|
if offset > 0:
|
|
params["offset"] = offset
|
|
return client.request("GET", "/api/user", params=params)
|
|
|
|
|
|
def metabase_get_user(client: MetabaseClient, user_id: int) -> dict:
|
|
"""Obtiene un usuario de Metabase por su ID.
|
|
|
|
Endpoint: GET /api/user/:id.
|
|
|
|
Args:
|
|
client: Cliente autenticado.
|
|
user_id: ID numerico del usuario.
|
|
|
|
Returns:
|
|
Dict con datos del usuario: id, email, first_name, last_name,
|
|
is_superuser, is_active, common_name, date_joined, last_login,
|
|
group_ids, locale.
|
|
|
|
Raises:
|
|
httpx.HTTPStatusError: 404 si el usuario no existe.
|
|
|
|
Example:
|
|
>>> user = metabase_get_user(client, 1)
|
|
>>> print(user["email"], user["is_superuser"])
|
|
"""
|
|
return client.request("GET", f"/api/user/{user_id}")
|
|
|
|
|
|
def metabase_create_user(
|
|
client: MetabaseClient,
|
|
first_name: str,
|
|
last_name: str,
|
|
email: str,
|
|
password: str = "",
|
|
group_ids: list[int] | None = None,
|
|
) -> dict:
|
|
"""Crea un nuevo usuario en Metabase.
|
|
|
|
Endpoint: POST /api/user. Requiere permisos de superusuario.
|
|
|
|
Args:
|
|
client: Cliente autenticado con permisos admin.
|
|
first_name: Nombre del usuario.
|
|
last_name: Apellido del usuario.
|
|
email: Email unico del usuario.
|
|
password: Password. Vacio = Metabase envia invitacion por email.
|
|
group_ids: IDs de grupos a asignar. None = solo grupo default.
|
|
|
|
Returns:
|
|
Dict con el usuario creado (mismos campos que metabase_get_user).
|
|
|
|
Raises:
|
|
httpx.HTTPStatusError: 400 si el email ya existe.
|
|
|
|
Example:
|
|
>>> user = metabase_create_user(client, "John", "Doe", "john@example.com", "pass123")
|
|
>>> print(user["id"])
|
|
"""
|
|
body: dict = {
|
|
"first_name": first_name,
|
|
"last_name": last_name,
|
|
"email": email,
|
|
}
|
|
if password:
|
|
body["password"] = password
|
|
if group_ids:
|
|
body["group_ids"] = group_ids
|
|
return client.request("POST", "/api/user", json=body)
|
|
|
|
|
|
def metabase_update_user(client: MetabaseClient, user_id: int, **fields) -> dict:
|
|
"""Actualiza campos de un usuario en Metabase.
|
|
|
|
Endpoint: PUT /api/user/:id. Requiere permisos de superusuario.
|
|
Solo se modifican los campos pasados como keyword arguments.
|
|
|
|
Args:
|
|
client: Cliente autenticado con permisos admin.
|
|
user_id: ID del usuario a actualizar.
|
|
**fields: Campos a actualizar. Validos:
|
|
first_name (str), last_name (str), email (str),
|
|
is_superuser (bool), group_ids (list[int]),
|
|
locale (str), login_attributes (dict).
|
|
|
|
Returns:
|
|
Dict con el usuario actualizado.
|
|
|
|
Example:
|
|
>>> user = metabase_update_user(client, 5, first_name="Jane", is_superuser=True)
|
|
>>> user = metabase_update_user(client, 5, group_ids=[1, 3, 5])
|
|
"""
|
|
return client.request("PUT", f"/api/user/{user_id}", json=fields)
|
|
|
|
|
|
def metabase_deactivate_user(client: MetabaseClient, user_id: int) -> None:
|
|
"""Desactiva (soft-delete) un usuario en Metabase.
|
|
|
|
Endpoint: DELETE /api/user/:id. Requiere permisos de superusuario.
|
|
El usuario no se elimina permanentemente, solo se marca como inactivo.
|
|
Para reactivar: PUT /api/user/:id/reactivate.
|
|
|
|
Args:
|
|
client: Cliente autenticado con permisos admin.
|
|
user_id: ID del usuario a desactivar.
|
|
|
|
Example:
|
|
>>> metabase_deactivate_user(client, 5)
|
|
>>> # Para ver desactivados: metabase_list_users(client, status="deactivated")
|
|
"""
|
|
client.request("DELETE", f"/api/user/{user_id}")
|