fn_registry/python/functions/metabase/cards.py

"""CRUD de cards/preguntas de Metabase y ejecucion de queries."""

from .client import MetabaseClient


def metabase_list_cards(
    client: MetabaseClient,
    filter: str = "",
    model_id: int = 0,
) -> list[dict]:
    """Lista preguntas/cards de Metabase con filtro opcional.

    Endpoint: GET /api/card. No tiene paginacion offset/limit.

    Args:
        client: Cliente autenticado.
        filter: "all", "mine", "fav", "archived", "recent", "popular",
                "database", "table". Vacio = todas.
        model_id: ID de database/tabla. Solo aplica con filter "database" o "table".

    Returns:
        Lista de dicts, cada uno con: id, name, description, display,
        collection_id, database_id, creator_id, archived, dataset_query.

    Example:
        >>> cards = metabase_list_cards(client, filter="mine")
        >>> for c in cards:
        ...     print(c["id"], c["name"], c["display"])
    """
    params = {}
    if filter:
        params["f"] = filter
    if model_id > 0:
        params["model_id"] = model_id
    return client.request("GET", "/api/card", params=params)


def metabase_get_card(client: MetabaseClient, card_id: int) -> dict:
    """Obtiene los detalles completos de una card/pregunta.

    Endpoint: GET /api/card/:id.

    Args:
        client: Cliente autenticado.
        card_id: ID de la card.

    Returns:
        Dict con: id, name, description, display, dataset_query,
        visualization_settings, collection_id, database_id, archived,
        creator, created_at, updated_at.

    Example:
        >>> card = metabase_get_card(client, 42)
        >>> print(card["name"], card["display"])
        >>> print(card["dataset_query"]["native"]["query"])  # SQL
    """
    return client.request("GET", f"/api/card/{card_id}")


def metabase_create_card(
    client: MetabaseClient,
    name: str,
    dataset_query: dict,
    display: str = "table",
    collection_id: int = 0,
    description: str = "",
) -> dict:
    """Crea una nueva card/pregunta en Metabase.

    Endpoint: POST /api/card.

    Args:
        client: Cliente autenticado.
        name: Nombre de la pregunta.
        dataset_query: Query de la card. Estructura:
            SQL nativo: {"database": 1, "type": "native", "native": {"query": "SELECT ..."}}
            MBQL: {"database": 1, "type": "query", "query": {"source-table": 4, ...}}
        display: Tipo de visualizacion: "table", "bar", "line", "pie", "scalar",
                 "area", "row", "combo", "funnel", "scatter", "waterfall", etc.
        collection_id: ID de coleccion destino. 0 = root.
        description: Descripcion opcional.

    Returns:
        Dict con la card creada.

    Example:
        >>> card = metabase_create_card(client, "Revenue by Month", {
        ...     "database": 1,
        ...     "type": "native",
        ...     "native": {"query": "SELECT date_trunc('month', created_at), SUM(total) FROM orders GROUP BY 1"},
        ... }, display="line", description="Monthly revenue trend")
    """
    body: dict = {
        "name": name,
        "dataset_query": dataset_query,
        "display": display,
        "visualization_settings": {},
    }
    if collection_id > 0:
        body["collection_id"] = collection_id
    if description:
        body["description"] = description
    return client.request("POST", "/api/card", json=body)


def metabase_update_card(client: MetabaseClient, card_id: int, **fields) -> dict:
    """Actualiza campos de una card/pregunta en Metabase.

    Endpoint: PUT /api/card/:id. Solo se modifican los campos pasados.

    Args:
        client: Cliente autenticado.
        card_id: ID de la card.
        **fields: Campos a actualizar. Validos:
            name (str), description (str), display (str),
            dataset_query (dict), visualization_settings (dict),
            collection_id (int), archived (bool),
            enable_embedding (bool), embedding_params (dict).

    Returns:
        Dict con la card actualizada.

    Example:
        >>> metabase_update_card(client, 42, name="Updated Name", archived=True)
        >>> metabase_update_card(client, 42, dataset_query={
        ...     "database": 1, "type": "native",
        ...     "native": {"query": "SELECT * FROM users LIMIT 100"},
        ... })
    """
    return client.request("PUT", f"/api/card/{card_id}", json=fields)


def metabase_delete_card(client: MetabaseClient, card_id: int) -> None:
    """Elimina permanentemente una card/pregunta.

    Endpoint: DELETE /api/card/:id. IRREVERSIBLE.
    Para soft-delete preferir: metabase_update_card(client, card_id, archived=True)

    Args:
        client: Cliente autenticado.
        card_id: ID de la card a eliminar.

    Example:
        >>> metabase_delete_card(client, 42)
        >>> # Preferir soft-delete: metabase_update_card(client, 42, archived=True)
    """
    client.request("DELETE", f"/api/card/{card_id}")


def metabase_execute_card(
    client: MetabaseClient,
    card_id: int,
    parameters: list[dict] | None = None,
) -> dict:
    """Ejecuta la query de una card/pregunta guardada.

    Endpoint: POST /api/card/:id/query.

    Args:
        client: Cliente autenticado.
        card_id: ID de la card a ejecutar.
        parameters: Parametros para queries parametrizadas. Cada parametro:
            {"type": "category", "target": ["variable", ["template-tag", "tag"]], "value": "val"}

    Returns:
        Dict con resultados:
        - status: "completed" o "failed"
        - row_count: numero de filas
        - running_time: tiempo en ms
        - data.columns: nombres de columnas
        - data.rows: filas de datos (lista de listas)
        - data.cols: metadata de columnas
        - data.native_form.query: SQL ejecutado

    Example:
        >>> result = metabase_execute_card(client, 42)
        >>> for row in result["data"]["rows"]:
        ...     print(row)
        >>> # Con parametros:
        >>> result = metabase_execute_card(client, 42, parameters=[
        ...     {"type": "category", "target": ["variable", ["template-tag", "status"]], "value": "active"},
        ... ])
    """
    body = {}
    if parameters:
        body["parameters"] = parameters
    return client.request("POST", f"/api/card/{card_id}/query", json=body or None)


def metabase_execute_query(
    client: MetabaseClient,
    database_id: int,
    sql: str,
    max_results: int = 0,
) -> dict:
    """Ejecuta una query SQL ad-hoc sin guardarla como card.

    Endpoint: POST /api/dataset. Util para exploracion rapida y consultas
    que no necesitan persistirse.

    Args:
        client: Cliente autenticado.
        database_id: ID de la database en Metabase.
        sql: Query SQL a ejecutar.
        max_results: Limite de filas. 0 = default 2000.

    Returns:
        Dict con misma estructura que metabase_execute_card:
        data.columns, data.rows, row_count, running_time, status.

    Example:
        >>> result = metabase_execute_query(client, 1, "SELECT * FROM users LIMIT 10")
        >>> print(f"{result['row_count']} filas en {result['running_time']}ms")
        >>> for row in result["data"]["rows"]:
        ...     print(row)
    """
    body: dict = {
        "database": database_id,
        "type": "native",
        "native": {"query": sql},
    }
    if max_results > 0:
        body["constraints"] = {
            "max-results": max_results,
            "max-results-bare-rows": max_results,
        }
    return client.request("POST", "/api/dataset", json=body)