Source Google Ads: make streams incremental

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

@@ -100,41 +100,53 @@
       "stream": {
         "name": "ad_group_ads",
         "json_schema": {},
-        "supported_sync_modes": ["full_refresh"],
-        "source_defined_primary_key": [["ad_group_ad.ad.id"]]
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "source_defined_primary_key": [["ad_group_ad.ad.id"]],
+        "default_cursor_field": ["segments.date"]
       },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite"
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite",
+      "cursor_field": ["segments.date"]
     },
     {
       "stream": {
         "name": "ad_groups",
         "json_schema": {},
-        "supported_sync_modes": ["full_refresh"],
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["segments.date"],
         "source_defined_primary_key": [["ad_group.id"]]
       },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite"
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite",
+      "cursor_field": ["segments.date"]
     },
     {
       "stream": {
         "name": "accounts",
         "json_schema": {},
-        "supported_sync_modes": ["full_refresh"],
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["segments.date"],
         "source_defined_primary_key": [["customer.id"]]
       },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite"
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite",
+      "cursor_field": ["segments.date"]
     },
     {
       "stream": {
         "name": "campaigns",
         "json_schema": {},
-        "supported_sync_modes": ["full_refresh"],
+        "supported_sync_modes": ["full_refresh", "incremental"],
+        "source_defined_cursor": true,
+        "default_cursor_field": ["segments.date"],
         "source_defined_primary_key": [["campaign.id"]]
       },
-      "sync_mode": "full_refresh",
-      "destination_sync_mode": "overwrite"
+      "sync_mode": "incremental",
+      "destination_sync_mode": "overwrite",
+      "cursor_field": ["segments.date"]
     },
     {
       "stream": {

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

@@ -64,6 +64,10 @@
     },
     "customer.tracking_url_template": {
       "type": ["null", "string"]
+    },
+    "segments.date": {
+      "type": ["null", "string"],
+      "format": "date"
     }
   }
 }

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

@@ -565,6 +565,10 @@
     },
     "ad_group_ad.status": {
       "type": ["null", "string"]
+    },
+    "segments.date": {
+      "type": ["null", "string"],
+      "format": "date"
     }
   }
 }

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

@@ -94,6 +94,10 @@
       "items": {
         "type": "string"
       }
+    },
+    "segments.date": {
+      "type": ["null", "string"],
+      "format": "date"
     }
   }
 }

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

@@ -238,6 +238,10 @@
     },
     "campaign.video_brand_safety_suitability": {
       "type": ["null", "string"]
+    },
+    "segments.date": {
+      "type": ["null", "string"],
+      "format": "date"
     }
   }
 }

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

@@ -46,8 +46,19 @@ def get_credentials(config: Mapping[str, Any]) -> Mapping[str, Any]:
         return credentials
 
     @staticmethod
-    def get_account_info(google_api) -> dict:
-        accounts_streams = Accounts(api=google_api)
+    def get_incremental_stream_config(google_api: GoogleAds, config: Mapping[str, Any], tz: Union[timezone, str] = "local"):
+        incremental_stream_config = dict(
+            api=google_api,
+            conversion_window_days=config["conversion_window_days"],
+            start_date=config["start_date"],
+            time_zone=tz,
+            end_date=config.get("end_date"),
+        )
+        return incremental_stream_config
+
+    def get_account_info(self, google_api: GoogleAds, config: Mapping[str, Any]) -> dict:
+        incremental_stream_config = self.get_incremental_stream_config(google_api, config)
+        accounts_streams = Accounts(**incremental_stream_config)
         for stream_slice in accounts_streams.stream_slices(sync_mode=SyncMode.full_refresh):
             return next(accounts_streams.read_records(sync_mode=SyncMode.full_refresh, stream_slice=stream_slice), {})
 
@@ -74,7 +85,7 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
         try:
             logger.info("Checking the config")
             google_api = GoogleAds(credentials=self.get_credentials(config), customer_id=config["customer_id"])
-            account_info = self.get_account_info(google_api)
+            account_info = self.get_account_info(google_api, config)
             is_manager_account = self.is_manager_account(account_info)
 
             # Check custom query request validity by sending metric request with non-existant time window
@@ -95,22 +106,15 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
 
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         google_api = GoogleAds(credentials=self.get_credentials(config), customer_id=config["customer_id"])
-        account_info = self.get_account_info(google_api)
+        account_info = self.get_account_info(google_api, config)
         time_zone = self.get_time_zone(account_info)
-        end_date = config.get("end_date")
-        incremental_stream_config = dict(
-            api=google_api,
-            conversion_window_days=config["conversion_window_days"],
-            start_date=config["start_date"],
-            time_zone=time_zone,
-            end_date=end_date,
-        )
+        incremental_stream_config = self.get_incremental_stream_config(google_api, config, tz=time_zone)
 
         streams = [
-            AdGroupAds(api=google_api),
-            AdGroups(api=google_api),
-            Accounts(api=google_api),
-            Campaigns(api=google_api),
+            AdGroupAds(**incremental_stream_config),
+            AdGroups(**incremental_stream_config),
+            Accounts(**incremental_stream_config),
+            Campaigns(**incremental_stream_config),
             ClickView(**incremental_stream_config),
         ]
 

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

@@ -215,36 +215,36 @@ def get_query(self, stream_slice: Mapping[str, Any] = None) -> str:
         return query
 
 
-class Accounts(GoogleAdsStream):
+class Accounts(IncrementalGoogleAdsStream):
     """
     Accounts stream: https://developers.google.com/google-ads/api/fields/v8/customer
     """
 
-    primary_key = "customer.id"
+    primary_key = ["customer.id", "segments.date"]
 
 
-class Campaigns(GoogleAdsStream):
+class Campaigns(IncrementalGoogleAdsStream):
     """
     Campaigns stream: https://developers.google.com/google-ads/api/fields/v8/campaign
     """
 
-    primary_key = "campaign.id"
+    primary_key = ["campaign.id", "segments.date"]
 
 
-class AdGroups(GoogleAdsStream):
+class AdGroups(IncrementalGoogleAdsStream):
     """
     AdGroups stream: https://developers.google.com/google-ads/api/fields/v8/ad_group
     """
 
-    primary_key = "ad_group.id"
+    primary_key = ["ad_group.id", "segments.date"]
 
 
-class AdGroupAds(GoogleAdsStream):
+class AdGroupAds(IncrementalGoogleAdsStream):
     """
     AdGroups stream: https://developers.google.com/google-ads/api/fields/v8/ad_group_ad
     """
 
-    primary_key = "ad_group_ad.ad.id"
+    primary_key = ["ad_group_ad.ad.id", "segments.date"]
 
 
 class AccountPerformanceReport(IncrementalGoogleAdsStream):