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