"""Cliente base para Google BigQuery.""" from dataclasses import dataclass, field from google.cloud import bigquery from google.oauth2 import service_account @dataclass class BQClient: """Cliente para Google BigQuery. Attributes: project_id: ID del proyecto GCP. _client: Cliente oficial de BigQuery (interno). """ project_id: str _client: bigquery.Client = field(repr=False) def close(self) -> None: """Cierra el cliente BigQuery.""" self._client.close() def __enter__(self): return self def __exit__(self, *args): self.close() def bq_auth(project_id: str = "", credentials_path: str = "") -> BQClient: """Autentica contra Google BigQuery. Tres modos de autenticacion: 1. ADC (Application Default Credentials): sin argumentos, usa gcloud auth 2. Service account JSON: con credentials_path 3. Proyecto explicito: con project_id (usa ADC para credenciales) Args: project_id: ID del proyecto GCP. Vacio = detectar de credenciales. credentials_path: Ruta a archivo JSON de service account. Vacio = ADC. Returns: BQClient autenticado listo para usar. Raises: google.auth.exceptions.DefaultCredentialsError: Si no hay credenciales configuradas. FileNotFoundError: Si credentials_path no existe. Example: >>> client = bq_auth() # ADC >>> client = bq_auth("my-project") # ADC con proyecto explicito >>> client = bq_auth(credentials_path="/path/to/sa.json") # Service account """ if credentials_path: creds = service_account.Credentials.from_service_account_file(credentials_path) proj = project_id or creds.project_id client = bigquery.Client(credentials=creds, project=proj) elif project_id: client = bigquery.Client(project=project_id) else: client = bigquery.Client() return BQClient(project_id=client.project, _client=client)