#12486 #49-alpha-beta-issues PR comment fixes

Author: davydov-d

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

@@ -91,7 +91,10 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
                 for query in config.get("custom_queries", []):
                     query = query.get("query")
                     if customer.is_manager_account and self.is_metrics_in_custom_query(query):
-                        return False, f"Metrics are not available for manager account. Check fields in your custom query: {query}"
+                        logger.warning(
+                            f"Metrics are not available for manager account {customer.id}. "
+                            f"Please remove metrics fields in your custom query: {query}."
+                        )
                     if CustomQuery.cursor_field in query:
                         return False, f"Custom query should not contain {CustomQuery.cursor_field}"
                     req_q = CustomQuery.insert_segments_date_expr(query, "1980-01-01", "1980-01-01")
@@ -109,37 +112,38 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         google_api = GoogleAds(credentials=self.get_credentials(config))
         accounts = self.get_account_info(google_api, config)
         customers = Customer.from_accounts(accounts)
-        incremental_stream_config = self.get_incremental_stream_config(google_api, config, customers)
+        non_manager_accounts = [customer for customer in customers if not customer.is_manager_account]
+        incremental_config = self.get_incremental_stream_config(google_api, config, customers)
+        non_manager_incremental_config = self.get_incremental_stream_config(google_api, config, non_manager_accounts)
         streams = [
-            AdGroupAds(**incremental_stream_config),
+            AdGroupAds(**incremental_config),
             AdGroupAdLabels(google_api, customers=customers),
-            AdGroups(**incremental_stream_config),
+            AdGroups(**incremental_config),
             AdGroupLabels(google_api, customers=customers),
-            Accounts(**incremental_stream_config),
-            Campaigns(**incremental_stream_config),
+            Accounts(**incremental_config),
+            Campaigns(**incremental_config),
             CampaignLabels(google_api, customers=customers),
-            ClickView(**incremental_stream_config),
-        ]
-        custom_query_streams = [
-            CustomQuery(custom_query_config=single_query_config, **incremental_stream_config)
-            for single_query_config in config.get("custom_queries", [])
+            ClickView(**incremental_config),
         ]
-        streams.extend(custom_query_streams)
-
         # Metrics streams cannot be requested for a manager account.
-        non_manager_accounts = [customer for customer in customers if not customer.is_manager_account]
         if non_manager_accounts:
-            incremental_stream_config["customers"] = non_manager_accounts
             streams.extend(
                 [
-                    UserLocationReport(**incremental_stream_config),
-                    AccountPerformanceReport(**incremental_stream_config),
-                    DisplayTopicsPerformanceReport(**incremental_stream_config),
-                    DisplayKeywordPerformanceReport(**incremental_stream_config),
-                    ShoppingPerformanceReport(**incremental_stream_config),
-                    AdGroupAdReport(**incremental_stream_config),
-                    GeographicReport(**incremental_stream_config),
-                    KeywordReport(**incremental_stream_config),
+                    UserLocationReport(**non_manager_incremental_config),
+                    AccountPerformanceReport(**non_manager_incremental_config),
+                    DisplayTopicsPerformanceReport(**non_manager_incremental_config),
+                    DisplayKeywordPerformanceReport(**non_manager_incremental_config),
+                    ShoppingPerformanceReport(**non_manager_incremental_config),
+                    AdGroupAdReport(**non_manager_incremental_config),
+                    GeographicReport(**non_manager_incremental_config),
+                    KeywordReport(**non_manager_incremental_config),
                 ]
             )
+        for single_query_config in config.get("custom_queries", []):
+            query = single_query_config.get("query")
+            if self.is_metrics_in_custom_query(query):
+                if non_manager_accounts:
+                    streams.append(CustomQuery(custom_query_config=single_query_config, **non_manager_incremental_config))
+                continue
+            streams.append(CustomQuery(custom_query_config=single_query_config, **incremental_config))
         return streams

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

@@ -9,6 +9,7 @@
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams import IncrementalMixin, Stream
 from google.ads.googleads.errors import GoogleAdsException
+from google.ads.googleads.v9.errors.types.authorization_error import AuthorizationErrorEnum
 from google.ads.googleads.v9.errors.types.request_error import RequestErrorEnum
 from google.ads.googleads.v9.services.services.google_ads_service.pagers import SearchPager
 
@@ -115,10 +116,10 @@ def read_records(self, sync_mode, stream_slice: Optional[Mapping[str, Any]] = No
             if not self.CATCH_API_ERRORS:
                 raise
             for error in exc.failure.errors:
-                if error.error_code.authentication_error or error.error_code.authorization_error or error.error_code.query_error:
+                if error.error_code.authorization_error == AuthorizationErrorEnum.AuthorizationError.CUSTOMER_NOT_ENABLED:
                     self.logger.error(error.message)
                     continue
-                # log and ignore only auth and query errors, otherwise - raise further
+                # log and ignore only CUSTOMER_NOT_ENABLED error, otherwise - raise further
                 raise
 
 

airbyte-integrations/connectors/source-google-ads/unit_tests/test_source.py

@@ -9,6 +9,7 @@
 from airbyte_cdk import AirbyteLogger
 from freezegun import freeze_time
 from google.ads.googleads.errors import GoogleAdsException
+from google.ads.googleads.v9.errors.types.authorization_error import AuthorizationErrorEnum
 from pendulum import today
 from source_google_ads.custom_query_stream import CustomQuery
 from source_google_ads.google_ads import GoogleAds
@@ -41,15 +42,15 @@ def mock(latest_record):
 
 @pytest.fixture
 def mocked_gads_api(mocker):
-    def mock(response=None, failure_msg="", error_type=""):
+    def mock(response=None, failure_code=1, failure_msg="", error_type=""):
         def side_effect_func():
-            raise make_google_ads_exception(failure_msg=failure_msg, error_type=error_type)
+            raise make_google_ads_exception(failure_code=failure_code, failure_msg=failure_msg, error_type=error_type)
             yield
 
         side_effect = []
         if response:
             side_effect.append(response)
-        if failure_msg or error_type:
+        if failure_msg or failure_code or error_type:
             side_effect.append(side_effect_func())
         mocker.patch("source_google_ads.google_ads.GoogleAds.send_request", side_effect=side_effect)
 
@@ -525,17 +526,17 @@ def test_invalid_custom_query_handled(mocked_gads_api, config):
 
 
 @pytest.mark.parametrize(
-    ("cls", "error", "raise_expected", "log_expected"),
+    ("cls", "error", "failure_code", "raise_expected", "log_expected"),
     (
-        (AdGroupLabels, "authentication_error", False, True),
-        (AdGroupLabels, "internal_error", True, False),
-        (ServiceAccounts, "authentication_error", True, False),
-        (ServiceAccounts, "internal_error", True, False),
+        (AdGroupLabels, "authorization_error", AuthorizationErrorEnum.AuthorizationError.CUSTOMER_NOT_ENABLED, False, True),
+        (AdGroupLabels, "internal_error", 1, True, False),
+        (ServiceAccounts, "authentication_error", 1, True, False),
+        (ServiceAccounts, "internal_error", 1, True, False),
     ),
 )
-def test_read_record_error_handling(config, customers, caplog, mocked_gads_api, cls, error, raise_expected, log_expected):
+def test_read_record_error_handling(config, customers, caplog, mocked_gads_api, cls, error, failure_code, raise_expected, log_expected):
     error_msg = "Some unexpected error"
-    mocked_gads_api(failure_msg=error_msg, error_type=error)
+    mocked_gads_api(failure_code=failure_code, failure_msg=error_msg, error_type=error)
     google_api = GoogleAds(credentials=config["credentials"])
     stream = cls(api=google_api, customers=customers)
     if raise_expected: