Skip to content

Modules

Bases: BaseClient

Source code in google_ads_report/google_ads_client.py
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class GoogleAdsApiClient(BaseClient):
    def __init__(self, credentials: dict = None, credentials_path: str = None, version: str = "v14"):
        """
        Args:
            credentials: A dictionary of credentials for the Google Ads API.
            credentials_path: The path to a JSON file containing credentials for the Google Ads API.
            version: The version of the Google Ads API to use.

        Returns:
            An instance of the GoogleAdsApiClient class.

        Author:
            minhpc@ikameglobal.com
        """
        super().__init__(version)

        if not credentials_path and not credentials:
            raise ValueError("Either credentials or credentials_path must be provided.")

        if not credentials:
            with open(credentials_path, 'r') as f:
                credentials = json.load(f)

        self.client = GoogleAdsClient.load_from_dict(credentials, version)
        self.ads_service = self.client.get_service("GoogleAdsService")
        self.version = version

    def get_response_batch(self, customer_id: str, query: str) -> Iterator[dict]:
        """
        Returns a generator of batches of results from the Google Ads API.

        Args:
            customer_id: The Google Ads customer ID.
            query: The query to run against the Google Ads API.

        Returns:
            A generator of batches of results from the Google Ads API.
        """
        stream = self.ads_service.search_stream(customer_id=customer_id,
                                                query=query)
        for batch in stream:
            batch_result = []
            for row in batch.results:
                row = json_format.MessageToDict(row)
                batch_result.append(row)
            yield batch_result

    def get_response(self, customer_id: str, query: str) -> list:
        """
        Returns a list of results from the Google Ads API.

        Args:
            customer_id: The Google Ads customer ID.
            query: The query to run against the Google Ads API.

        Returns:
            A list of results from the Google Ads API.
        """
        stream = self.ads_service.search_stream(customer_id=customer_id,
                                                query=query)
        result = []
        for batch in stream:
            for row in batch.results:
                row = json_format.MessageToDict(row)
                result.append(row)
        return result

Parameters:

Name Type Description Default
credentials dict

A dictionary of credentials for the Google Ads API.

None
credentials_path str

The path to a JSON file containing credentials for the Google Ads API.

None
version str

The version of the Google Ads API to use.

'v14'

Returns:

Type Description

An instance of the GoogleAdsApiClient class.

Author

minhpc@ikameglobal.com

Source code in google_ads_report/google_ads_client.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def __init__(self, credentials: dict = None, credentials_path: str = None, version: str = "v14"):
    """
    Args:
        credentials: A dictionary of credentials for the Google Ads API.
        credentials_path: The path to a JSON file containing credentials for the Google Ads API.
        version: The version of the Google Ads API to use.

    Returns:
        An instance of the GoogleAdsApiClient class.

    Author:
        minhpc@ikameglobal.com
    """
    super().__init__(version)

    if not credentials_path and not credentials:
        raise ValueError("Either credentials or credentials_path must be provided.")

    if not credentials:
        with open(credentials_path, 'r') as f:
            credentials = json.load(f)

    self.client = GoogleAdsClient.load_from_dict(credentials, version)
    self.ads_service = self.client.get_service("GoogleAdsService")
    self.version = version

Returns a list of results from the Google Ads API.

Parameters:

Name Type Description Default
customer_id str

The Google Ads customer ID.

required
query str

The query to run against the Google Ads API.

required

Returns:

Type Description
list

A list of results from the Google Ads API.

Source code in google_ads_report/google_ads_client.py
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def get_response(self, customer_id: str, query: str) -> list:
    """
    Returns a list of results from the Google Ads API.

    Args:
        customer_id: The Google Ads customer ID.
        query: The query to run against the Google Ads API.

    Returns:
        A list of results from the Google Ads API.
    """
    stream = self.ads_service.search_stream(customer_id=customer_id,
                                            query=query)
    result = []
    for batch in stream:
        for row in batch.results:
            row = json_format.MessageToDict(row)
            result.append(row)
    return result

Returns a generator of batches of results from the Google Ads API.

Parameters:

Name Type Description Default
customer_id str

The Google Ads customer ID.

required
query str

The query to run against the Google Ads API.

required

Returns:

Type Description
Iterator[dict]

A generator of batches of results from the Google Ads API.

Source code in google_ads_report/google_ads_client.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def get_response_batch(self, customer_id: str, query: str) -> Iterator[dict]:
    """
    Returns a generator of batches of results from the Google Ads API.

    Args:
        customer_id: The Google Ads customer ID.
        query: The query to run against the Google Ads API.

    Returns:
        A generator of batches of results from the Google Ads API.
    """
    stream = self.ads_service.search_stream(customer_id=customer_id,
                                            query=query)
    for batch in stream:
        batch_result = []
        for row in batch.results:
            row = json_format.MessageToDict(row)
            batch_result.append(row)
        yield batch_result