dataforge/fn_registry
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat(metabase): nuevos modulos — snippets, notifications, dashboard_filters

Browse Source 
Tres modulos nuevos con funciones CRUD completas:
- snippets: list, get, create, update, archive (SQL reutilizable)
- notifications: list, create_card_alert, create_dashboard_subscription, update, delete
- dashboard_filters: add_dashboard_filter (parameter_mappings sobre cards existentes)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
egutierrez
2026-04-14 19:03:08 +02:00
parent 7081c3b4d1
commit 4299482b75
14 changed files with 1105 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/functions/metabase/dashboard_filters.py
+90
View File
@@ -0,0 +1,90 @@
"""Helpers para añadir filtros a dashboards de Metabase."""
import uuid as _uuid
from .client import MetabaseClient
from .metabase_update_dashboard_safe import metabase_update_dashboard_safe
def metabase_add_dashboard_filter(
client: MetabaseClient,
dashboard_id: int,
name: str,
slug: str,
filter_type: str,
card_mappings: list[dict],
*,
default: str | None = None,
) -> dict:
"""Añade un filtro a un dashboard y lo conecta a las cards indicadas.
Crea un parametro a nivel de dashboard y actualiza los parameter_mappings
de cada dashcard indicada. Usa metabase_update_dashboard_safe internamente.
Args:
client: Cliente autenticado.
dashboard_id: ID del dashboard.
name: Nombre visible del filtro (ej: "Fecha desde").
slug: Slug URL-friendly (ej: "fecha_desde").
filter_type: Tipo del filtro. Comunes: "string/=", "number/=",
"date/single", "date/range", "date/relative", "category".
card_mappings: Lista de dicts con la conexión filtro→card. Formato:
[{"card_id": 42, "template_tag": "fecha_desde"}, ...]
Cada dict indica qué card y qué template-tag SQL debe recibir el valor.
default: Valor por defecto del filtro (opcional).
Returns:
Dict con resultado de metabase_update_dashboard_safe.
Example:
>>> result = metabase_add_dashboard_filter(
... client, 887, "Fecha desde", "fecha_desde", "string/=",
... [{"card_id": 7711, "template_tag": "fecha_desde"},
... {"card_id": 7719, "template_tag": "fecha_desde"}],
... )
"""
from .dashboards import metabase_get_dashboard
# 1. Leer dashboard actual
dash = metabase_get_dashboard(client, dashboard_id)
current_params = dash.get("parameters", [])
current_dashcards = dash.get("dashcards", [])
# 2. Crear el parameter
param_id = _uuid.uuid4().hex[:8]
new_param = {
"id": param_id,
"type": filter_type,
"name": name,
"slug": slug,
}
if default is not None:
new_param["default"] = default
updated_params = current_params + [new_param]
# 3. Build card_id → template_tag mapping
tag_by_card = {m["card_id"]: m["template_tag"] for m in card_mappings}
# 4. Update dashcards with parameter_mappings
updated_dashcards = []
for dc in current_dashcards:
dc_card_id = dc.get("card_id")
if dc_card_id in tag_by_card:
mappings = list(dc.get("parameter_mappings", []))
mappings.append({
"parameter_id": param_id,
"card_id": dc_card_id,
"target": ["variable", ["template-tag", tag_by_card[dc_card_id]]],
})
dc_copy = {**dc, "parameter_mappings": mappings}
updated_dashcards.append(dc_copy)
else:
updated_dashcards.append(dc)
# 5. Update via safe function
return metabase_update_dashboard_safe(
client, dashboard_id,
dashcards_update=updated_dashcards,
extra_fields={"parameters": updated_params},
)
python/functions/metabase/metabase_add_dashboard_filter.md
+76
View File
@@ -0,0 +1,76 @@
---
name: metabase_add_dashboard_filter
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_add_dashboard_filter(client: MetabaseClient, dashboard_id: int, name: str, slug: str, filter_type: str, card_mappings: list[dict], *, default: str | None = None) -> dict"
description: "Añade un filtro a un dashboard de Metabase y lo conecta a las cards indicadas. Crea el parameter a nivel de dashboard y actualiza los parameter_mappings de cada dashcard."
tags: [metabase, dashboard, filter, parameter, create, api, python]
uses_functions:
- metabase_update_dashboard_safe_py_infra
- metabase_get_dashboard_py_infra
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [uuid]
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: dashboard_id
desc: "ID del dashboard al que se añade el filtro"
- name: name
desc: "nombre visible del filtro en la UI de Metabase (ej: 'Fecha desde')"
- name: slug
desc: "slug URL-friendly del filtro, unico dentro del dashboard (ej: 'fecha_desde')"
- name: filter_type
desc: "tipo del filtro: 'string/=', 'number/=', 'date/single', 'date/range', 'date/relative', 'category'"
- name: card_mappings
desc: "lista de dicts [{card_id: int, template_tag: str}] conectando el filtro a template-tags SQL de las cards"
- name: default
desc: "valor por defecto del filtro; None = sin default"
output: "dict con resultado de metabase_update_dashboard_safe: {added, updated, removed, response}"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/dashboard_filters.py"
---
## Ejemplo
```python
result = metabase_add_dashboard_filter(
client,
dashboard_id=887,
name="Fecha desde",
slug="fecha_desde",
filter_type="string/=",
card_mappings=[
{"card_id": 7711, "template_tag": "fecha_desde"},
{"card_id": 7719, "template_tag": "fecha_desde"},
],
)
# Con valor por defecto
result = metabase_add_dashboard_filter(
client,
dashboard_id=887,
name="Estado",
slug="estado",
filter_type="category",
card_mappings=[{"card_id": 7711, "template_tag": "estado"}],
default="activo",
)
```
## Notas
Realiza 2 requests internamente via metabase_update_dashboard_safe (GET + PUT).
El `param_id` se genera como UUID hex de 8 caracteres — formato que usa la UI de Metabase.
`card_mappings` solo conecta cards que ya existen como dashcards en el dashboard. Cards no presentes en el dashboard se ignoran silenciosamente.
Para filtros de tipo date, los valores en `default` deben seguir el formato de Metabase: "2024-01-01" para `date/single`, "2024-01-01~2024-12-31" para `date/range`.
python/functions/metabase/metabase_archive_snippet.md
+45
View File
@@ -0,0 +1,45 @@
---
name: metabase_archive_snippet
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_archive_snippet(client: MetabaseClient, snippet_id: int) -> dict"
description: "Archiva un SQL snippet en Metabase. Wrapper sobre metabase_update_snippet con archived=True."
tags: [metabase, snippet, archive, api, python]
uses_functions: [metabase_update_snippet_py_infra]
uses_types: [MetabaseClient_go_infra]
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [httpx]
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: snippet_id
desc: "ID numerico del snippet a archivar"
output: "dict: snippet con archived=True y updated_at actualizado"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/snippets.py"
---
## Ejemplo
```python
# Archivar snippet obsoleto
result = metabase_archive_snippet(client, 42)
print(result["archived"]) # True
# Listar snippets activos despues de archivar
active = metabase_list_snippets(client, archived=False)
assert all(not s["archived"] for s in active)
```
## Notas
Los snippets archivados no aparecen en el autocomplete de queries nativas en el editor de Metabase.
Las cards que ya referencian el snippet siguen funcionando correctamente despues de archivar.
Para desarchivar, usar metabase_update_snippet(client, snippet_id, archived=False).
python/functions/metabase/metabase_create_card_alert.md
+59
View File
@@ -0,0 +1,59 @@
---
name: metabase_create_card_alert
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_create_card_alert(client: MetabaseClient, card_id: int, cron_schedule: str, recipients: list[dict], send_condition: str = 'has_result', send_once: bool = False) -> dict"
description: "Crea una alerta sobre los resultados de una card en Metabase. Envia email segun cron cuando la card cumple la condicion (has_result, goal_above, goal_below). Endpoint: POST /api/notification."
tags: [metabase, notification, alert, card, create, api, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: card_id
desc: "ID de la card que dispara la alerta"
- name: cron_schedule
desc: "expresion cron de 5 campos (ej: '0 9 * * 1' = lunes 9am, '0 8 * * 1-5' = lun-vie 8am)"
- name: recipients
desc: "lista de destinatarios: usuario Metabase {'type': 'notification-recipient/user', 'user_id': N} o email externo {'type': 'notification-recipient/raw-value', 'details': {'email': 'x@y.com'}}"
- name: send_condition
desc: "condicion que dispara el envio: 'has_result' (tiene filas), 'goal_above' (supera goal), 'goal_below' (cae bajo goal)"
- name: send_once
desc: "si True, se envia una sola vez y se desactiva automaticamente"
output: "dict: notificacion creada con id, active, payload_type, payload, subscriptions, handlers, created_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/notifications.py"
---
## Ejemplo
```python
# Alerta: enviar email los lunes 9am si la card tiene resultados
alert = metabase_create_card_alert(
client,
card_id=7711,
cron_schedule="0 9 * * 1",
recipients=[
{"type": "notification-recipient/user", "user_id": 1},
{"type": "notification-recipient/raw-value", "details": {"email": "team@aurgi.com"}},
],
send_condition="has_result",
)
print(alert["id"], alert["active"])
```
## Notas
Requiere Metabase 0.57+. Reemplaza el antiguo /api/pulse.
El campo `subscriptions` acepta solo el tipo `notification-subscription/cron`.
Para alertas de goal configurar previamente el goal en la visualizacion de la card.
`send_once=True` es util para alertas puntuales que no deben repetirse.
python/functions/metabase/metabase_create_dashboard_subscription.md
+53
View File
@@ -0,0 +1,53 @@
---
name: metabase_create_dashboard_subscription
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_create_dashboard_subscription(client: MetabaseClient, dashboard_id: int, cron_schedule: str, recipients: list[dict]) -> dict"
description: "Crea una suscripcion periodica a un dashboard de Metabase. Envia el dashboard completo por email segun el cron configurado. Endpoint: POST /api/notification."
tags: [metabase, notification, subscription, dashboard, create, api, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: dashboard_id
desc: "ID del dashboard a enviar periodicamente"
- name: cron_schedule
desc: "expresion cron de 5 campos (ej: '0 8 * * 1-5' = lun-vie 8am, '0 9 * * 1' = lunes 9am)"
- name: recipients
desc: "lista de destinatarios: usuario Metabase {'type': 'notification-recipient/user', 'user_id': N} o email externo {'type': 'notification-recipient/raw-value', 'details': {'email': 'x@y.com'}}"
output: "dict: suscripcion creada con id, active, payload_type, payload, subscriptions, handlers, created_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/notifications.py"
---
## Ejemplo
```python
# Suscripcion: enviar dashboard de ventas cada lunes a las 8am
sub = metabase_create_dashboard_subscription(
client,
dashboard_id=42,
cron_schedule="0 8 * * 1",
recipients=[
{"type": "notification-recipient/user", "user_id": 5},
{"type": "notification-recipient/raw-value", "details": {"email": "gerencia@aurgi.com"}},
],
)
print(sub["id"], sub["payload"]["dashboard_id"])
```
## Notas
Requiere Metabase 0.57+. Reemplaza el antiguo /api/pulse.
El dashboard se envia completo con todas sus cards renderizadas.
Para suscripciones diarias laborales usar cron "0 8 * * 1-5".
python/functions/metabase/metabase_create_snippet.md
+70
View File
@@ -0,0 +1,70 @@
---
name: metabase_create_snippet
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_create_snippet(client: MetabaseClient, name: str, content: str, description: str = \"\", collection_id: int = 0) -> dict"
description: "Crea un nuevo SQL snippet reutilizable en Metabase. El snippet se referencia en queries nativas con {{snippet: nombre}}."
tags: [metabase, snippet, create, api, python]
uses_functions: []
uses_types: [MetabaseClient_go_infra]
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [httpx]
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: name
desc: "nombre del snippet; se usa en queries con la sintaxis {{snippet: nombre}}"
- name: content
desc: "SQL del snippet: puede ser una CTE completa, subquery o cualquier fragmento SQL reutilizable"
- name: description
desc: "descripcion opcional del proposito del snippet"
- name: collection_id
desc: "ID de la coleccion donde guardar el snippet; 0 para la raiz (Our analytics)"
output: "dict: snippet creado con id asignado por Metabase, name, content, description, collection_id, created_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/snippets.py"
---
## Ejemplo
```python
# Crear snippet con CTE reutilizable
snippet = metabase_create_snippet(
client,
"supply_orders_cte",
"""
WITH supply_full AS (
SELECT so.id, so.service_request_id,
sr.channel_id, ch.name AS channel_name
FROM supply_orders so
LEFT JOIN service_requests sr ON so.service_request_id = sr.id
LEFT JOIN channels ch ON sr.channel_id = ch.id
)
""",
description="CTE base de supply_orders con 6 JOINs para cruce con NAV",
)
print(snippet["id"], snippet["name"])
# Usar el snippet en una card
from metabase.cards import metabase_create_card
card = metabase_create_card(client, "Revenue", {
"database": 6,
"type": "native",
"native": {
"query": "{{snippet: supply_orders_cte}} SELECT channel_name, COUNT(*) FROM supply_full GROUP BY 1"
}
})
```
## Notas
El campo `name` debe ser unico en Metabase — el servidor retorna 400 si ya existe un snippet con ese nombre.
Si `collection_id` es 0 o no se provee, el snippet se guarda en la raiz.
Solo se envian al body los campos no vacios/cero para evitar conflictos con defaults del servidor.
python/functions/metabase/metabase_delete_notification.md
+44
View File
@@ -0,0 +1,44 @@
---
name: metabase_delete_notification
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_delete_notification(client: MetabaseClient, notification_id: int) -> None"
description: "Elimina una notificacion de Metabase (alerta de card o suscripcion de dashboard). Operacion irreversible. Endpoint: DELETE /api/notification/:id."
tags: [metabase, notification, delete, api, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: notification_id
desc: "ID de la notificacion a eliminar"
output: "None"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/notifications.py"
---
## Ejemplo
```python
# Eliminar una alerta por su ID
metabase_delete_notification(client, 99)
# Patron: listar y eliminar todas las alertas de una card
alerts = metabase_list_notifications(client, card_id=7711)
for a in alerts:
metabase_delete_notification(client, a["id"])
```
## Notas
La operacion es irreversible. Para desactivar temporalmente usar metabase_update_notification con active=False.
Elimina tanto alertas de cards (notification/card) como suscripciones de dashboards (notification/dashboard).
python/functions/metabase/metabase_get_snippet.md
+41
View File
@@ -0,0 +1,41 @@
---
name: metabase_get_snippet
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_get_snippet(client: MetabaseClient, snippet_id: int) -> dict"
description: "Obtiene un SQL snippet de Metabase por su ID. Endpoint: GET /api/native-query-snippet/:id."
tags: [metabase, snippet, get, api, python]
uses_functions: []
uses_types: [MetabaseClient_go_infra]
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [httpx]
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: snippet_id
desc: "ID numerico del snippet a obtener"
output: "dict: snippet completo con id, name, content, description, collection_id, creator_id, archived, created_at, updated_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/snippets.py"
---
## Ejemplo
```python
snippet = metabase_get_snippet(client, 42)
print(snippet["name"])
print(snippet["content"])
print(snippet["description"])
```
## Notas
Retorna 404 si el snippet no existe.
Util para verificar el contenido actual antes de hacer un update.
python/functions/metabase/metabase_list_notifications.md
+49
View File
@@ -0,0 +1,49 @@
---
name: metabase_list_notifications
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_list_notifications(client: MetabaseClient, card_id: int = 0, dashboard_id: int = 0) -> list[dict]"
description: "Lista notificaciones de Metabase (alertas y suscripciones). Filtra opcionalmente por card_id o dashboard_id. Endpoint: GET /api/notification."
tags: [metabase, notification, list, api, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: card_id
desc: "si > 0, filtra alertas de tipo notification/card asociadas a esta card"
- name: dashboard_id
desc: "si > 0, filtra suscripciones de tipo notification/dashboard asociadas a este dashboard"
output: "list[dict]: lista de notificaciones con id, active, payload_type, payload, subscriptions, handlers, created_at, updated_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/notifications.py"
---
## Ejemplo
```python
# Todas las notificaciones
notifs = metabase_list_notifications(client)
# Solo alertas de una card especifica
alerts = metabase_list_notifications(client, card_id=7711)
for a in alerts:
print(a["id"], a["payload_type"], a["active"])
# Solo suscripciones de un dashboard
subs = metabase_list_notifications(client, dashboard_id=42)
```
## Notas
Requiere Metabase 0.57+. En versiones anteriores usar /api/pulse (deprecado).
Sin filtros retorna todas las notificaciones accesibles por el usuario autenticado.
python/functions/metabase/metabase_list_snippets.md
+45
View File
@@ -0,0 +1,45 @@
---
name: metabase_list_snippets
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_list_snippets(client: MetabaseClient, archived: bool = False) -> list[dict]"
description: "Lista SQL snippets reutilizables de Metabase. Un snippet se referencia en queries con {{snippet: nombre}}."
tags: [metabase, snippet, list, api, python]
uses_functions: []
uses_types: [MetabaseClient_go_infra]
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [httpx]
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: archived
desc: "si True, incluye snippets archivados (default False)"
output: "list[dict]: snippets con id, name, content, description, collection_id, creator_id, archived, created_at, updated_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/snippets.py"
---
## Ejemplo
```python
snippets = metabase_list_snippets(client)
for s in snippets:
print(s["name"], len(s["content"]))
# Incluir archivados
all_snippets = metabase_list_snippets(client, archived=True)
active = [s for s in all_snippets if not s["archived"]]
```
## Notas
Si `archived=True` agrega el query param `?archived=true`.
Retorna lista directa de todos los snippets accesibles.
Los snippets archivados no aparecen en el autocomplete de queries de Metabase.
python/functions/metabase/metabase_update_notification.md
+64
View File
@@ -0,0 +1,64 @@
---
name: metabase_update_notification
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_update_notification(client: MetabaseClient, notification_id: int, **fields) -> dict"
description: "Actualiza una notificacion existente de Metabase (alerta o suscripcion). Permite modificar active, handlers, subscriptions o payload. Endpoint: PUT /api/notification/:id."
tags: [metabase, notification, update, api, python]
uses_functions: []
uses_types: []
returns: []
returns_optional: false
error_type: "error_go_core"
imports: []
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: notification_id
desc: "ID de la notificacion a modificar"
- name: "**fields"
desc: "campos a actualizar: active (bool), handlers (list[dict]), subscriptions (list[dict]), payload (dict)"
output: "dict: notificacion actualizada con id, active, payload_type, payload, subscriptions, handlers, updated_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/notifications.py"
---
## Ejemplo
```python
# Desactivar una alerta
updated = metabase_update_notification(client, 99, active=False)
print(updated["active"]) # False
# Cambiar destinatarios
updated = metabase_update_notification(
client,
99,
handlers=[{
"channel_type": "channel/email",
"recipients": [
{"type": "notification-recipient/user", "user_id": 2},
],
}],
)
# Cambiar cron a diario a las 7am
updated = metabase_update_notification(
client,
99,
subscriptions=[{
"type": "notification-subscription/cron",
"cron_schedule": "0 7 * * *",
}],
)
```
## Notas
Solo se envian los campos que se pasan como kwargs.
Para reactivar una alerta previamente desactivada usar active=True.
python/functions/metabase/metabase_update_snippet.md
+56
View File
@@ -0,0 +1,56 @@
---
name: metabase_update_snippet
kind: function
lang: py
domain: infra
version: "1.0.0"
purity: impure
signature: "def metabase_update_snippet(client: MetabaseClient, snippet_id: int, **fields) -> dict"
description: "Actualiza campos de un SQL snippet en Metabase. Acepta name, content, description, collection_id, archived via **fields."
tags: [metabase, snippet, update, api, python]
uses_functions: []
uses_types: [MetabaseClient_go_infra]
returns: []
returns_optional: false
error_type: "error_go_core"
imports: [httpx]
params:
- name: client
desc: "instancia autenticada de MetabaseClient"
- name: snippet_id
desc: "ID numerico del snippet a actualizar"
- name: "**fields"
desc: "campos a modificar: name (str), content (str), description (str), collection_id (int), archived (bool)"
output: "dict: snippet actualizado con todos los campos incluyendo updated_at"
tested: false
tests: []
test_file_path: ""
file_path: "python/functions/metabase/snippets.py"
---
## Ejemplo
```python
# Actualizar solo el contenido SQL
updated = metabase_update_snippet(
client, 42,
content="WITH supply_full AS (SELECT * FROM supply_orders)",
description="Version simplificada sin JOINs",
)
print(updated["updated_at"])
# Renombrar
metabase_update_snippet(client, 42, name="supply_orders_simple_cte")
# Mover a otra coleccion
metabase_update_snippet(client, 42, collection_id=7)
# Archivar (o usar metabase_archive_snippet)
metabase_update_snippet(client, 42, archived=True)
```
## Notas
Solo los campos reconocidos (name, content, description, collection_id, archived) se incluyen en el body.
Campos desconocidos en **fields se ignoran silenciosamente para evitar errores 400.
Para archivar en una sola llamada, preferir metabase_archive_snippet.
python/functions/metabase/notifications.py
+243
View File
@@ -0,0 +1,243 @@
"""CRUD de notificaciones de Metabase (alertas y suscripciones).
Dos tipos:
- Card alerts (notification/card): alertas cuando una card tiene resultados,
supera un goal, etc. Cron configurable.
- Dashboard subscriptions (notification/dashboard): envio periodico de un
dashboard completo por email/slack.
API: /api/notification (Metabase 0.57+, reemplaza el antiguo /api/pulse).
"""
from .client import MetabaseClient
def metabase_list_notifications(
client: MetabaseClient,
card_id: int = 0,
dashboard_id: int = 0,
) -> list[dict]:
"""Lista notificaciones activas de Metabase.
Endpoint: GET /api/notification.
Permite filtrar por card o por dashboard. Si no se pasan filtros,
retorna todas las notificaciones accesibles por el usuario autenticado.
Args:
client: Cliente autenticado.
card_id: Si > 0, filtra notificaciones de tipo notification/card
asociadas a esta card.
dashboard_id: Si > 0, filtra notificaciones de tipo
notification/dashboard asociadas a este dashboard.
Returns:
Lista de dicts con: id, active, payload_type, payload,
subscriptions, handlers, created_at, updated_at.
Example:
>>> # Todas las notificaciones
>>> notifs = metabase_list_notifications(client)
>>> # Solo alertas de una card especifica
>>> alerts = metabase_list_notifications(client, card_id=7711)
>>> for a in alerts:
... print(a["id"], a["payload_type"], a["active"])
"""
params: dict = {}
if card_id > 0:
params["card_id"] = card_id
if dashboard_id > 0:
params["dashboard_id"] = dashboard_id
return client.request("GET", "/api/notification", params=params)
def metabase_create_card_alert(
client: MetabaseClient,
card_id: int,
cron_schedule: str,
recipients: list[dict],
send_condition: str = "has_result",
send_once: bool = False,
) -> dict:
"""Crea una alerta sobre los resultados de una card.
Endpoint: POST /api/notification.
Envia un email cuando la card cumple la condicion especificada
(tiene resultados, supera un goal, etc.) segun el cron configurado.
Args:
client: Cliente autenticado.
card_id: ID de la card que dispara la alerta.
cron_schedule: Expresion cron de 5 campos (ej: "0 9 * * 1" = lunes 9am).
recipients: Lista de destinatarios. Cada dict puede ser:
- Usuario Metabase: {"type": "notification-recipient/user", "user_id": 1}
- Email externo: {"type": "notification-recipient/raw-value",
"details": {"email": "x@y.com"}}
send_condition: Condicion que dispara el envio:
"has_result" (tiene filas), "goal_above" (supera goal),
"goal_below" (cae bajo goal). Default: "has_result".
send_once: Si True, se envia una sola vez y se desactiva.
Default: False.
Returns:
Dict con la notificacion creada: id, active, payload_type,
payload, subscriptions, handlers, created_at.
Example:
>>> alert = metabase_create_card_alert(
... client,
... card_id=7711,
... cron_schedule="0 9 * * 1",
... recipients=[
... {"type": "notification-recipient/user", "user_id": 1},
... {"type": "notification-recipient/raw-value",
... "details": {"email": "team@aurgi.com"}},
... ],
... send_condition="has_result",
... )
>>> print(alert["id"], alert["active"])
"""
body = {
"payload_type": "notification/card",
"payload": {
"card_id": card_id,
"send_condition": send_condition,
"send_once": send_once,
},
"subscriptions": [
{
"type": "notification-subscription/cron",
"cron_schedule": cron_schedule,
}
],
"handlers": [
{
"channel_type": "channel/email",
"recipients": recipients,
}
],
"active": True,
}
return client.request("POST", "/api/notification", json=body)
def metabase_create_dashboard_subscription(
client: MetabaseClient,
dashboard_id: int,
cron_schedule: str,
recipients: list[dict],
) -> dict:
"""Crea una suscripcion periodica a un dashboard.
Endpoint: POST /api/notification.
Envia el dashboard completo por email segun el cron configurado.
Args:
client: Cliente autenticado.
dashboard_id: ID del dashboard a enviar.
cron_schedule: Expresion cron de 5 campos (ej: "0 8 * * 1-5" = lun-vie 8am).
recipients: Lista de destinatarios. Cada dict puede ser:
- Usuario Metabase: {"type": "notification-recipient/user", "user_id": 1}
- Email externo: {"type": "notification-recipient/raw-value",
"details": {"email": "x@y.com"}}
Returns:
Dict con la suscripcion creada: id, active, payload_type,
payload, subscriptions, handlers, created_at.
Example:
>>> sub = metabase_create_dashboard_subscription(
... client,
... dashboard_id=42,
... cron_schedule="0 8 * * 1",
... recipients=[
... {"type": "notification-recipient/user", "user_id": 5},
... ],
... )
>>> print(sub["id"], sub["payload"]["dashboard_id"])
"""
body = {
"payload_type": "notification/dashboard",
"payload": {
"dashboard_id": dashboard_id,
},
"subscriptions": [
{
"type": "notification-subscription/cron",
"cron_schedule": cron_schedule,
}
],
"handlers": [
{
"channel_type": "channel/email",
"recipients": recipients,
}
],
"active": True,
}
return client.request("POST", "/api/notification", json=body)
def metabase_update_notification(
client: MetabaseClient,
notification_id: int,
**fields,
) -> dict:
"""Actualiza una notificacion existente (alerta o suscripcion).
Endpoint: PUT /api/notification/:id.
Permite modificar campos como active, handlers, subscriptions,
payload, etc. Solo se envian los campos proporcionados.
Args:
client: Cliente autenticado.
notification_id: ID de la notificacion a modificar.
**fields: Campos a actualizar. Ejemplos:
active=False para desactivar.
handlers=[...] para cambiar destinatarios.
subscriptions=[...] para cambiar el cron.
Returns:
Dict con la notificacion actualizada: id, active, payload_type,
payload, subscriptions, handlers, updated_at.
Example:
>>> # Desactivar una alerta
>>> updated = metabase_update_notification(client, 99, active=False)
>>> # Cambiar destinatarios
>>> updated = metabase_update_notification(
... client,
... 99,
... handlers=[{
... "channel_type": "channel/email",
... "recipients": [
... {"type": "notification-recipient/user", "user_id": 2},
... ],
... }],
... )
>>> print(updated["active"])
"""
return client.request("PUT", f"/api/notification/{notification_id}", json=fields)
def metabase_delete_notification(
client: MetabaseClient,
notification_id: int,
) -> None:
"""Elimina una notificacion (alerta o suscripcion) de Metabase.
Endpoint: DELETE /api/notification/:id.
La operacion es irreversible. Afecta tanto a alertas de cards
como a suscripciones de dashboards.
Args:
client: Cliente autenticado.
notification_id: ID de la notificacion a eliminar.
Returns:
None.
Example:
>>> metabase_delete_notification(client, 99)
>>> # La notificacion ya no existe
"""
client.request("DELETE", f"/api/notification/{notification_id}")
python/functions/metabase/snippets.py
+170
View File
@@ -0,0 +1,170 @@
"""CRUD de SQL snippets de Metabase.
Un snippet es un fragmento SQL reutilizable que se referencia en queries
nativas con la sintaxis {{snippet: nombre}}. Ideal para CTEs comunes
que se repiten en multiples cards.
"""
from .client import MetabaseClient
def metabase_list_snippets(client: MetabaseClient, archived: bool = False) -> list[dict]:
"""Lista SQL snippets de Metabase.
Endpoint: GET /api/native-query-snippet.
Args:
client: Cliente autenticado.
archived: Si True, incluye snippets archivados.
Returns:
Lista de dicts con: id, name, content, description, collection_id,
creator_id, archived, created_at, updated_at.
Example:
>>> snippets = metabase_list_snippets(client)
>>> for s in snippets:
... print(s["name"], len(s["content"]))
"""
params = {}
if archived:
params["archived"] = "true"
return client.request("GET", "/api/native-query-snippet", params=params)
def metabase_get_snippet(client: MetabaseClient, snippet_id: int) -> dict:
"""Obtiene un SQL snippet de Metabase por su ID.
Endpoint: GET /api/native-query-snippet/:id.
Args:
client: Cliente autenticado.
snippet_id: ID numerico del snippet.
Returns:
Dict con campos completos del snippet: id, name, content, description,
collection_id, creator_id, archived, created_at, updated_at.
Raises:
httpx.HTTPStatusError: 404 si el snippet no existe.
Example:
>>> snippet = metabase_get_snippet(client, 42)
>>> print(snippet["name"], snippet["content"])
"""
return client.request("GET", f"/api/native-query-snippet/{snippet_id}")
def metabase_create_snippet(
client: MetabaseClient,
name: str,
content: str,
description: str = "",
collection_id: int = 0,
) -> dict:
"""Crea un nuevo SQL snippet en Metabase.
Endpoint: POST /api/native-query-snippet.
Args:
client: Cliente autenticado.
name: Nombre del snippet. Se usa en queries con {{snippet: nombre}}.
content: SQL del snippet. Puede ser una CTE completa, una subquery,
o cualquier fragmento SQL reutilizable.
description: Descripcion opcional del snippet.
collection_id: ID de la coleccion donde guardar el snippet.
Si es 0, se guarda en la raiz (Our analytics).
Returns:
Dict con el snippet creado, incluyendo el campo "id" asignado
por Metabase.
Raises:
httpx.HTTPStatusError: 400 si el nombre ya existe o el SQL es invalido.
Example:
>>> snippet = metabase_create_snippet(
... client,
... "supply_orders_cte",
... '''WITH supply_full AS (
... SELECT so.id, so.service_request_id
... FROM supply_orders so
... LEFT JOIN service_requests sr ON so.service_request_id = sr.id
... )''',
... description="CTE base de supply_orders con JOINs para cruce con NAV",
... )
>>> print(snippet["id"], snippet["name"])
>>> # Usar en una card:
>>> # "{{snippet: supply_orders_cte}} SELECT ... FROM supply_full"
"""
body: dict = {"name": name, "content": content}
if description:
body["description"] = description
if collection_id:
body["collection_id"] = collection_id
return client.request("POST", "/api/native-query-snippet", json=body)
def metabase_update_snippet(
client: MetabaseClient,
snippet_id: int,
**fields,
) -> dict:
"""Actualiza campos de un SQL snippet en Metabase.
Endpoint: PUT /api/native-query-snippet/:id.
Args:
client: Cliente autenticado.
snippet_id: ID numerico del snippet a actualizar.
**fields: Campos a modificar. Campos validos:
- name (str): Nuevo nombre del snippet.
- content (str): Nuevo SQL del snippet.
- description (str): Nueva descripcion.
- collection_id (int): Nueva coleccion.
- archived (bool): True para archivar, False para desarchivar.
Returns:
Dict con el snippet actualizado.
Raises:
httpx.HTTPStatusError: 404 si el snippet no existe.
httpx.HTTPStatusError: 400 si los campos son invalidos.
Example:
>>> updated = metabase_update_snippet(
... client, 42,
... content="WITH supply_full AS (SELECT * FROM supply_orders)",
... description="Version simplificada",
... )
>>> print(updated["updated_at"])
"""
valid_fields = {"name", "content", "description", "collection_id", "archived"}
body = {k: v for k, v in fields.items() if k in valid_fields}
return client.request("PUT", f"/api/native-query-snippet/{snippet_id}", json=body)
def metabase_archive_snippet(client: MetabaseClient, snippet_id: int) -> dict:
"""Archiva un SQL snippet en Metabase.
Wrapper sobre metabase_update_snippet con archived=True.
Los snippets archivados no aparecen en el autocomplete de queries
pero sus referencias existentes siguen funcionando.
Endpoint: PUT /api/native-query-snippet/:id.
Args:
client: Cliente autenticado.
snippet_id: ID numerico del snippet a archivar.
Returns:
Dict con el snippet archivado.
Raises:
httpx.HTTPStatusError: 404 si el snippet no existe.
Example:
>>> result = metabase_archive_snippet(client, 42)
>>> print(result["archived"]) # True
"""
return metabase_update_snippet(client, snippet_id, archived=True)