🎉 Source Google Ads: add new stream 'click_view' (#6684)

* fetch gclid from clickview stream

* resolve comments and update google ads document

* add document url in stream.py file

* fixed date range issue and added campaign name in json schema

Co-authored-by: blotouta2 <ankur@blotout.io>

Author: yevhenii-ldv

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/253487c0-2246-43ba-a21f-5116b20a2c50.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "253487c0-2246-43ba-a21f-5116b20a2c50",
   "name": "Google Ads",
   "dockerRepository": "airbyte/source-google-ads",
-  "dockerImageTag": "0.1.11",
+  "dockerImageTag": "0.1.15",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/google-ads"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -33,7 +33,7 @@
 - sourceDefinitionId: 253487c0-2246-43ba-a21f-5116b20a2c50
   name: Google Ads
   dockerRepository: airbyte/source-google-ads
-  dockerImageTag: 0.1.14
+  dockerImageTag: 0.1.15
   documentationUrl: https://docs.airbyte.io/integrations/sources/google-ads
   sourceType: api
 - sourceDefinitionId: ef69ef6e-aa7f-4af1-a01d-ef775033524e

airbyte-integrations/connectors/source-google-ads/Dockerfile

@@ -13,5 +13,5 @@ RUN pip install .
 
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.14
+LABEL io.airbyte.version=0.1.15
 LABEL io.airbyte.name=airbyte/source-google-ads

airbyte-integrations/connectors/source-google-ads/integration_tests/configured_catalog.json

@@ -24,6 +24,18 @@
       "destination_sync_mode": "overwrite",
       "cursor_field": ["segments.date"]
     },
+    {
+      "stream": {
+        "name": "click_view",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["segments.date"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite",
+      "cursor_field": ["segments.date"]
+    },
     {
       "stream": {
         "name": "display_keyword_performance_report",

airbyte-integrations/connectors/source-google-ads/integration_tests/configured_catalog_without_empty_streams.json

@@ -54,6 +54,18 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "click_view",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["segments.date"]
+      },
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite",
+      "cursor_field": ["segments.date"]
+    },
     {
       "stream": {
         "name": "campaigns",

airbyte-integrations/connectors/source-google-ads/source_google_ads/google_ads.py

@@ -6,6 +6,7 @@
 from enum import Enum
 from typing import Any, List, Mapping
 
+import pendulum
 from google.ads.googleads.client import GoogleAdsClient
 from google.ads.googleads.v8.services.types.google_ads_service import GoogleAdsRow, SearchGoogleAdsResponse
 from proto.marshal.collections import Repeated, RepeatedComposite
@@ -21,6 +22,7 @@
     "display_topics_performance_report": "topic_view",
     "shopping_performance_report": "shopping_performance_view",
     "user_location_report": "user_location_view",
+    "click_view": "click_view",
 }
 
 
@@ -79,8 +81,10 @@ def convert_schema_into_query(
         query_template = f"SELECT {fields} FROM {from_category} "
 
         if cursor_field:
-            # Fix issue 5411: Make date_start and date_end inclusive.
-            query_template += f"WHERE {cursor_field} >= '{from_date}' AND {cursor_field} <= '{to_date}' ORDER BY {cursor_field} ASC"
+            end_date_inclusive = "<=" if (pendulum.parse(to_date) - pendulum.parse(from_date)).days > 1 else "<"
+            query_template += (
+                f"WHERE {cursor_field} >= '{from_date}' AND {cursor_field} {end_date_inclusive} '{to_date}' ORDER BY {cursor_field} ASC"
+            )
 
         return query_template
 

airbyte-integrations/connectors/source-google-ads/source_google_ads/schemas/click_view.json

@@ -0,0 +1,24 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "type": "object",
+  "properties": {
+    "click_view.gclid": {
+      "type": ["null", "string"]
+    },
+    "campaign.id": {
+      "type": ["null", "integer"]
+    },
+    "ad_group.id": {
+      "type": ["null", "integer"]
+    },
+    "segments.date": {
+      "type": ["null", "string"]
+    },
+    "customer.id": {
+      "type": ["null", "integer"]
+    },
+    "campaign.name": {
+      "type": ["null", "string"]
+    }
+  }
+}

airbyte-integrations/connectors/source-google-ads/source_google_ads/source.py

@@ -20,6 +20,7 @@
     AdGroupAds,
     AdGroups,
     Campaigns,
+    ClickView,
     DisplayKeywordPerformanceReport,
     DisplayTopicsPerformanceReport,
     ShoppingPerformanceReport,
@@ -74,4 +75,5 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             AdGroups(api=google_api),
             Accounts(api=google_api),
             Campaigns(api=google_api),
+            ClickView(**incremental_stream_config),
         ] + custom_query_streams

airbyte-integrations/connectors/source-google-ads/source_google_ads/streams.py

@@ -12,7 +12,9 @@
 from .google_ads import GoogleAds
 
 
-def chunk_date_range(start_date: str, conversion_window: int, field: str, end_date: str = None) -> Iterable[Mapping[str, any]]:
+def chunk_date_range(
+    start_date: str, conversion_window: int, field: str, end_date: str = None, time_unit: str = "months", days_of_data_storage: int = None
+) -> Iterable[Mapping[str, any]]:
     """
     Passing optional parameter end_date for testing
     Returns a list of the beginning and ending timetsamps of each month between the start date and now.
@@ -22,6 +24,10 @@ def chunk_date_range(start_date: str, conversion_window: int, field: str, end_da
     end_date = pendulum.parse(end_date) if end_date else pendulum.now()
     start_date = pendulum.parse(start_date)
 
+    # For some metrics we can only get data not older than N days, it is Google Ads policy
+    if days_of_data_storage:
+        start_date = max(start_date, pendulum.now().subtract(days=days_of_data_storage - conversion_window))
+
     # As in to return some state when state in abnormal
     if start_date > end_date:
         return [{field: start_date.to_date_string()}]
@@ -32,7 +38,7 @@ def chunk_date_range(start_date: str, conversion_window: int, field: str, end_da
     # Each stream_slice contains the beginning and ending timestamp for a 24 hour period
     while start_date < end_date:
         intervals.append({field: start_date.to_date_string()})
-        start_date = start_date.add(months=1)
+        start_date = start_date.add(**{time_unit: 1})
 
     return intervals
 
@@ -55,8 +61,10 @@ def read_records(self, sync_mode, stream_slice: Mapping[str, Any] = None, **kwar
 
 
 class IncrementalGoogleAdsStream(GoogleAdsStream, ABC):
+    days_of_data_storage = None
     cursor_field = "segments.date"
     primary_key = None
+    time_unit = "months"
 
     def __init__(self, start_date: str, conversion_window_days: int, **kwargs):
         self.conversion_window_days = conversion_window_days
@@ -67,16 +75,22 @@ def stream_slices(self, stream_state: Mapping[str, Any] = None, **kwargs) -> Ite
         stream_state = stream_state or {}
         start_date = stream_state.get(self.cursor_field) or self._start_date
 
-        return chunk_date_range(start_date=start_date, conversion_window=self.conversion_window_days, field=self.cursor_field)
+        return chunk_date_range(
+            start_date=start_date,
+            conversion_window=self.conversion_window_days,
+            field=self.cursor_field,
+            time_unit=self.time_unit,
+            days_of_data_storage=self.days_of_data_storage,
+        )
 
     @staticmethod
-    def get_date_params(stream_slice: Mapping[str, Any], cursor_field: str, end_date: pendulum.datetime = None):
+    def get_date_params(stream_slice: Mapping[str, Any], cursor_field: str, end_date: pendulum.datetime = None, time_unit: str = "months"):
         end_date = end_date or pendulum.yesterday()
         start_date = pendulum.parse(stream_slice.get(cursor_field))
         if start_date > pendulum.now():
             return start_date.to_date_string(), start_date.add(days=1).to_date_string()
 
-        end_date = min(end_date, pendulum.parse(stream_slice.get(cursor_field)).add(months=1))
+        end_date = min(end_date, pendulum.parse(stream_slice.get(cursor_field)).add(**{time_unit: 1}))
 
         # Fix issue #4806, start date should always be lower than end date.
         if start_date.add(days=1).date() >= end_date.date():
@@ -100,7 +114,7 @@ def get_updated_state(self, current_stream_state: MutableMapping[str, Any], late
         return current_stream_state
 
     def get_query(self, stream_slice: Mapping[str, Any] = None) -> str:
-        start_date, end_date = self.get_date_params(stream_slice, self.cursor_field)
+        start_date, end_date = self.get_date_params(stream_slice, self.cursor_field, time_unit=self.time_unit)
         query = GoogleAds.convert_schema_into_query(
             schema=self.get_json_schema(), report_name=self.name, from_date=start_date, to_date=end_date, cursor_field=self.cursor_field
         )
@@ -179,3 +193,12 @@ class UserLocationReport(IncrementalGoogleAdsStream):
     UserLocationReport stream: https://developers.google.com/google-ads/api/fields/v8/user_location_view
     Google Ads API field mapping: https://developers.google.com/google-ads/api/docs/migration/mapping#geo_performance
     """
+
+
+class ClickView(IncrementalGoogleAdsStream):
+    """
+    ClickView stream: https://developers.google.com/google-ads/api/reference/rpc/v8/ClickView
+    """
+
+    time_unit = "days"
+    days_of_data_storage = 90

docs/integrations/sources/google-ads.md

@@ -22,6 +22,7 @@ This source is capable of syncing the following tables and their data:
 * [ad_group_ads](https://developers.google.com/google-ads/api/fields/v8/ad_group_ad)
 * [ad_groups](https://developers.google.com/google-ads/api/fields/v8/ad_group)
 * [campaigns](https://developers.google.com/google-ads/api/fields/v8/campaign)
+* [click_view](https://developers.google.com/google-ads/api/reference/rpc/v8/ClickView)
 
 #### Report Tables
 * [account_performance_report](https://developers.google.com/google-ads/api/docs/migration/mapping#account_performance)
@@ -30,6 +31,8 @@ This source is capable of syncing the following tables and their data:
 * [display_topics_report](https://developers.google.com/google-ads/api/docs/migration/mapping#display_topics_performance)
 * [shopping_performance_report](https://developers.google.com/google-ads/api/docs/migration/mapping#shopping_performance)
 
+**Note**: Due to constraints from the Google Ads API, the `click_view` stream retrieves data one day at a time and can only retrieve data newer than 90 days ago
+
 ## Getting Started (Airbyte-Cloud)
 
 1. Click `Authenticate your Google Ads account` to sign in with Google and authorize your account.
@@ -91,6 +94,8 @@ This source is constrained by whatever API limits are set for the Google Ads tha
 
 | Version | Date | Pull Request | Subject |
 | :------ | :--------  | :-----       | :------ |
+| `0.1.15` | 2021-10-07 | [6684](https://github.com/airbytehq/airbyte/pull/6684) | Add new stream `click_view` |
+| `0.1.14` | 2021-10-01 | [6565](https://github.com/airbytehq/airbyte/pull/6565) | Fix OAuth Spec File |
 | `0.1.13` | 2021-09-27 | [6458](https://github.com/airbytehq/airbyte/pull/6458) | Update OAuth Spec File |
 | `0.1.11` | 2021-09-22 | [#6373](https://github.com/airbytehq/airbyte/pull/6373) | Fix inconsistent segments.date field type across all streams |
 | `0.1.10` | 2021-09-13 | [#6022](https://github.com/airbytehq/airbyte/pull/6022) | Annotate Oauth2 flow initialization parameters in connector spec |