fn_registry/python/functions/metabase/notifications.py

"""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}")