Merge pull request #15 from google-marketing-solutions/dev

Dev

Author: maximilianw-google

py/common/google_ads_client.py

@@ -14,7 +14,7 @@
 
 """The Google Ads client."""
 
-from typing import Any, Optional, Union
+from typing import Any, Literal, Optional, Union
 from common import api_utils
 
 GOOGLE_ADS_API_BASE_URL = 'https://googleads.googleapis.com'
@@ -270,3 +270,83 @@ def get_active_keywords_for_account(self, customer_id: str) -> list[Any]:
         api_version=self.api_version, customer_id=customer_id
     )
     return api_utils.send_api_request(url, payload, self._get_http_header())
+
+  def get_extensions_for_campaigns(
+      self,
+      customer_id: str,
+      campaign_ids: Optional[list[Union[int, str]]] = None,
+  ) -> list[Any]:
+    """Gets extension assets for campaigns and ad groups.
+
+    Gets existing structured snippets, callouts and sitelink extensions.
+
+    Args:
+      customer_id: The Google Ads customer id.
+      campaign_ids: A list of Google Ads campaign ids.
+
+    Returns:
+      The API response object containing a list of extensions.
+    """
+    results = []
+    for level in ['ad_group', 'campaign']:
+      results.extend(
+          self._get_extensions_for_campaigns_by_level(
+              customer_id, level, campaign_ids
+          )
+      )
+    return results
+
+  def _get_extensions_for_campaigns_by_level(
+      self,
+      customer_id: str,
+      level: Literal['ad_group', 'campaign'],
+      campaign_ids: Optional[list[Union[int, str]]] = None,
+  ) -> list[Any]:
+    """Gets extension assets for ad groups or campaigns.
+
+    Gets existing structured snippets, callouts and sitelink extensions.
+
+    Args:
+      customer_id: The Google Ads account id.
+      level: The level to get extensions for - ad_group or campaign.
+      campaign_ids: A list of Google Ads campaign ids.
+
+    Returns:
+      The API response object containing a list of extensions.
+    """
+    ad_group_name_col = ', ad_group.name' if level == 'ad_group' else ''
+    campaign_col = (
+        'ad_group.campaign' if level == 'ad_group' else 'campaign.resource_name'
+    )
+    query = f"""
+        SELECT
+            campaign.name,
+            {campaign_col},
+            asset.type,
+            asset.structured_snippet_asset.header,
+            asset.structured_snippet_asset.values,
+            asset.callout_asset.callout_text,
+            asset.sitelink_asset.description1,
+            asset.sitelink_asset.description2,
+            asset.sitelink_asset.link_text,
+            asset.final_urls,
+            {level}_asset.status{ad_group_name_col}
+        FROM
+          {level}_asset
+        WHERE
+          {level}_asset.field_type IN (
+              'STRUCTURED_SNIPPET', 'SITELINK', 'CALLOUT')
+        """
+    if campaign_ids:
+      campaign_ids_str = ', '.join(
+          [
+              f"'customers/{customer_id}/campaigns/{elem}'"
+              for elem in campaign_ids
+          ]
+      )
+      query += f' AND {campaign_col} IN ({campaign_ids_str})'
+    payload = {'query': ' '.join(query.split())}
+    url = SEARCH_URL.format(
+        api_version=self.api_version, customer_id=customer_id
+    )
+    return api_utils.send_api_request(url, payload, self._get_http_header())

py/common/google_ads_client_test.py

@@ -361,6 +361,94 @@ def test_not_success_http_code_in_response_raises_error(self, mock_requests):
     with self.assertRaises(requests.exceptions.HTTPError):
       self.client.get_accounts(_FAKE_CUSTOMER_ID)
 
+  @parameterized.named_parameters(
+      {
+          'testcase_name': 'without_campaign_ids',
+          'params': {
+              'customer_id': 1234567890,
+              'campaign_ids': None,
+          },
+          'expected_query_part_ad_group': '',
+          'expected_query_part_campaign': '',
+      },
+      {
+          'testcase_name': 'with_campaign_ids',
+          'params': {
+              'customer_id': 1234567890,
+              'campaign_ids': [1],
+          },
+          'expected_query_part_ad_group': (
+              " AND ad_group.campaign IN ('customers/1234567890/campaigns/1')"
+          ),
+          'expected_query_part_campaign': (
+              ' AND campaign.resource_name IN'
+              " ('customers/1234567890/campaigns/1')"
+          ),
+      },
+  )
+  @requests_mock.Mocker()
+  def test_get_extensions_for_campaigns(
+      self,
+      mock_requests,
+      params,
+      expected_query_part_ad_group,
+      expected_query_part_campaign,
+  ):
+    mock_response = _FAKE_RESPONSE
+    mock_requests.post(
+        requests_mock.ANY, json=mock_response, headers=_FAKE_HEADERS
+    )
+    query_ad_group = """
+        SELECT
+            campaign.name,
+            ad_group.campaign,
+            asset.type,
+            asset.structured_snippet_asset.header,
+            asset.structured_snippet_asset.values,
+            asset.callout_asset.callout_text,
+            asset.sitelink_asset.description1,
+            asset.sitelink_asset.description2,
+            asset.sitelink_asset.link_text,
+            asset.final_urls,
+            ad_group_asset.status, ad_group.name
+        FROM
+          ad_group_asset
+        WHERE
+          ad_group_asset.field_type IN (
+              'STRUCTURED_SNIPPET', 'SITELINK', 'CALLOUT')
+        """ + expected_query_part_ad_group
+    query_campaign = """
+        SELECT
+            campaign.name,
+            campaign.resource_name,
+            asset.type,
+            asset.structured_snippet_asset.header,
+            asset.structured_snippet_asset.values,
+            asset.callout_asset.callout_text,
+            asset.sitelink_asset.description1,
+            asset.sitelink_asset.description2,
+            asset.sitelink_asset.link_text,
+            asset.final_urls,
+            campaign_asset.status
+        FROM
+          campaign_asset
+        WHERE
+          campaign_asset.field_type IN (
+              'STRUCTURED_SNIPPET', 'SITELINK', 'CALLOUT')
+        """ + expected_query_part_campaign
+    expected_request_ad_group = _expected_request_from_query(query_ad_group)
+    expected_request_campaign = _expected_request_from_query(query_campaign)
+    self.client.get_extensions_for_campaigns(**params)
+
+    self.assertEqual(
+        _request_record_asdict(mock_requests.request_history[0]),
+        expected_request_ad_group,
+    )
+    self.assertEqual(
+        _request_record_asdict(mock_requests.request_history[1]),
+        expected_request_campaign,
+    )
+
 
 if __name__ == '__main__':
   absltest.main()