feat: add BigQuery Python functions and BQClient type

Funciones CRUD completas para BigQuery: auth, datasets, tables, queries, jobs, routines, load/export. Tipo BQClient como wrapper del SDK oficial.
2026-04-07 18:45:02 +02:00
parent c311623a76
commit 690e68a542
33 changed files with 2720 additions and 0 deletions
@@ -0,0 +1,134 @@
+"""Gestion de routines (UDFs y stored procedures) en Google BigQuery."""
+
+from .client import BQClient
+from google.cloud import bigquery
+
+
+def bq_create_routine(
+    client: BQClient,
+    dataset_id: str,
+    routine_id: str,
+    body: str,
+    routine_type: str = "SCALAR_FUNCTION",
+    language: str = "SQL",
+    arguments: list[dict] | None = None,
+    return_type: str = "",
+    description: str = "",
+) -> dict:
+    """Crea una routine (UDF o stored procedure) en BigQuery.
+
+    Args:
+        client: Cliente autenticado.
+        dataset_id: ID del dataset donde crear la routine.
+        routine_id: ID/nombre de la routine.
+        body: Cuerpo de la routine (SQL, JavaScript o Python).
+        routine_type: Tipo: "SCALAR_FUNCTION", "TABLE_VALUED_FUNCTION", "PROCEDURE".
+        language: Lenguaje: "SQL", "JAVASCRIPT", "PYTHON".
+        arguments: Lista de argumentos, cada uno: {"name": "x", "data_type": "STRING"}.
+        return_type: Tipo de retorno para funciones (ej: "STRING", "INT64"). No aplica a procedures.
+        description: Descripcion opcional.
+
+    Returns:
+        Dict con: routine_id, dataset_id, project, routine_type, language, body, created, modified.
+
+    Raises:
+        google.api_core.exceptions.Conflict: Si la routine ya existe.
+
+    Example:
+        >>> bq_create_routine(client, "analytics", "double_value",
+        ...     body="x * 2",
+        ...     arguments=[{"name": "x", "data_type": "INT64"}],
+        ...     return_type="INT64")
+    """
+    ref = bigquery.RoutineReference.from_string(
+        f"{client.project_id}.{dataset_id}.{routine_id}"
+    )
+    routine = bigquery.Routine(ref)
+    routine.type_ = routine_type
+    routine.language = language
+    routine.body = body
+    if description:
+        routine.description = description
+    if arguments:
+        routine.arguments = [
+            bigquery.RoutineArgument(
+                name=arg["name"],
+                data_type=bigquery.StandardSqlDataType(
+                    type_kind=getattr(
+                        bigquery.StandardSqlTypeNames, arg["data_type"]
+                    )
+                ),
+            )
+            for arg in arguments
+        ]
+    if return_type:
+        routine.return_type = bigquery.StandardSqlDataType(
+            type_kind=getattr(bigquery.StandardSqlTypeNames, return_type)
+        )
+    created = client._client.create_routine(routine)
+    return _routine_to_dict(created)
+
+
+def bq_list_routines(client: BQClient, dataset_id: str) -> list[dict]:
+    """Lista routines de un dataset.
+
+    Args:
+        client: Cliente autenticado.
+        dataset_id: ID del dataset.
+
+    Returns:
+        Lista de dicts con: routine_id, dataset_id, project, routine_type, language.
+
+    Example:
+        >>> routines = bq_list_routines(client, "analytics")
+        >>> for r in routines:
+        ...     print(r["routine_id"], r["routine_type"], r["language"])
+    """
+    ref = f"{client.project_id}.{dataset_id}"
+    return [
+        {
+            "routine_id": r.routine_id,
+            "dataset_id": dataset_id,
+            "project": client.project_id,
+            "routine_type": r.type_,
+            "language": r.language,
+        }
+        for r in client._client.list_routines(ref)
+    ]
+
+
+def bq_delete_routine(
+    client: BQClient,
+    dataset_id: str,
+    routine_id: str,
+) -> None:
+    """Elimina una routine.
+
+    Args:
+        client: Cliente autenticado.
+        dataset_id: ID del dataset.
+        routine_id: ID de la routine a eliminar.
+
+    Raises:
+        google.api_core.exceptions.NotFound: Si la routine no existe.
+
+    Example:
+        >>> bq_delete_routine(client, "analytics", "double_value")
+    """
+    ref = f"{client.project_id}.{dataset_id}.{routine_id}"
+    client._client.delete_routine(ref)
+
+
+def _routine_to_dict(routine) -> dict:
+    """Convierte un objeto Routine del SDK a dict plano."""
+    return {
+        "routine_id": routine.routine_id,
+        "dataset_id": routine.dataset_id,
+        "project": routine.project,
+        "routine_type": routine.type_,
+        "language": routine.language,
+        "body": routine.body,
+        "description": routine.description or "",
+        "created": routine.created.isoformat() if routine.created else None,
+        "modified": routine.modified.isoformat() if routine.modified else None,
+    }