"""Google Ads API client wrapper."""
from typing import Any
from google.ads.googleads.client import GoogleAdsClient
from google.ads.googleads.errors import GoogleAdsException
from .config import settings
class GoogleAdsClientWrapper:
"""Wrapper for Google Ads API client with authentication handling."""
_instance: "GoogleAdsClientWrapper | None" = None
_client: GoogleAdsClient | None = None
def __new__(cls) -> "GoogleAdsClientWrapper":
"""Singleton pattern to reuse client connection."""
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def _get_credentials_dict(self) -> dict[str, Any]:
"""Build credentials dictionary for Google Ads client."""
credentials = {
"client_id": settings.client_id,
"client_secret": settings.client_secret,
"developer_token": settings.developer_token,
"refresh_token": settings.refresh_token,
"use_proto_plus": True,
}
if settings.login_customer_id:
credentials["login_customer_id"] = settings.login_customer_id
return credentials
@property
def client(self) -> GoogleAdsClient:
"""Get or create the Google Ads client."""
if self._client is None:
if not settings.is_configured:
raise ValueError(
"Google Ads credentials not configured. "
"Please set environment variables or create a .env file."
)
self._client = GoogleAdsClient.load_from_dict(self._get_credentials_dict())
return self._client
@property
def customer_id(self) -> str:
"""Get the configured customer ID."""
return settings.customer_id
def get_service(self, service_name: str) -> Any:
"""Get a Google Ads service by name.
Args:
service_name: Name of the service (e.g., "GoogleAdsService", "CampaignService")
Returns:
The requested service client.
"""
return self.client.get_service(service_name)
def search(self, query: str, customer_id: str | None = None) -> list[Any]:
"""Execute a GAQL query and return results.
Args:
query: Google Ads Query Language query string.
customer_id: Optional customer ID (defaults to configured ID).
Returns:
List of result rows.
"""
ga_service = self.get_service("GoogleAdsService")
cid = customer_id or self.customer_id
try:
response = ga_service.search(customer_id=cid, query=query)
return list(response)
except GoogleAdsException as ex:
# Re-raise with more context
raise GoogleAdsException(
error=ex.error,
call=ex.call,
failure=ex.failure,
request_id=ex.request_id,
) from ex
# Global client instance
ads_client = GoogleAdsClientWrapper()
