🎉Source Bing Ads: Add Report streams

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/47f25999-dd5e-4636-8c39-e7cea2453331.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "47f25999-dd5e-4636-8c39-e7cea2453331",
   "name": "Bing Ads",
   "dockerRepository": "airbyte/source-bing-ads",
-  "dockerImageTag": "0.1.0",
+  "dockerImageTag": "0.1.1",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/bing-ads"
 }

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

@@ -435,7 +435,7 @@
 - sourceDefinitionId: 47f25999-dd5e-4636-8c39-e7cea2453331
   name: Bing Ads
   dockerRepository: airbyte/source-bing-ads
-  dockerImageTag: 0.1.0
+  dockerImageTag: 0.1.1
   documentationUrl: https://docs.airbyte.io/integrations/sources/bing-ads
 - sourceDefinitionId: 59c5501b-9f95-411e-9269-7143c939adbd
   name: BigCommerce

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

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.0
+LABEL io.airbyte.version=0.1.1
 LABEL io.airbyte.name=airbyte/source-bing-ads

airbyte-integrations/connectors/source-bing-ads/acceptance-test-config.yml

@@ -12,7 +12,7 @@ tests:
   basic_read:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/configured_catalog.json"
-  # connector doesn't have incremental streams
+  # incremental test doesn't work if a single stream has multiple states
   #incremental:
   #  - config_path: "secrets/config.json"
   #    configured_catalog_path: "integration_tests/configured_catalog.json"

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

@@ -35,6 +35,66 @@
       },
       "sync_mode": "full_refresh",
       "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "campaign_performance_report",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["TimePeriod"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "budget_summary_report",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["Date"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "ad_performance_report",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["TimePeriod"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "ad_group_performance_report",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["TimePeriod"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "account_performance_report",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["TimePeriod"],
+      "destination_sync_mode": "append"
+    },
+    {
+      "stream": {
+        "name": "keyword_performance_report",
+        "json_schema": {},
+        "supported_sync_modes": ["incremental", "full_refresh"]
+      },
+      "sync_mode": "incremental",
+      "cursor_field": ["TimePeriod"],
+      "destination_sync_mode": "append"
     }
   ]
 }

airbyte-integrations/connectors/source-bing-ads/integration_tests/invalid_config.json

@@ -1,9 +1,11 @@
 {
-  "accounts": { "type": "all" },
+  "accounts": { "selection_strategy": "all" },
   "user_id": "2222",
   "customer_id": "1111",
   "developer_token": "asgag4gwag3",
   "refresh_token": "as2Ggas23gsa236gasgaskjfhas7i8ygf78as7osa7gy87asg8as7tg6as",
   "client_secret": "1234",
-  "client_id": "123"
+  "client_id": "123",
+  "reports_start_date": "2018-11-13",
+  "report_aggregation": { "value": "Monthly" }
 }

airbyte-integrations/connectors/source-bing-ads/integration_tests/state.json

@@ -0,0 +1,22 @@
+{
+  "keyword_performance_report": {
+    "180278106": {
+      "TimePeriod": 1627820152
+    }
+  },
+  "budget_summary_report": {
+    "180278106": {
+      "Date": 1627800152
+    }
+  },
+  "ad_performance_report": {
+    "180278106": {
+      "TimePeriod": 1627795152
+    }
+  },
+  "campaign_performance_report": {
+    "180278106": {
+      "TimePeriod": 1727810152
+    }
+  }
+}

airbyte-integrations/connectors/source-bing-ads/source_bing_ads/client.py

@@ -23,15 +23,17 @@
 #
 
 import sys
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from functools import lru_cache
 from typing import Any, Iterator, Mapping, Optional
 
 import backoff
+import pendulum
 from airbyte_cdk.logger import AirbyteLogger
 from bingads.authorization import AuthorizationData, OAuthTokens, OAuthWebAuthCodeGrant
 from bingads.service_client import ServiceClient
 from bingads.util import errorcode_of_exception
+from bingads.v13.reporting.reporting_service_manager import ReportingServiceManager
 from suds import WebFault, sudsobject
 
 
@@ -42,11 +44,15 @@ class Client:
     # retry on: rate limit errors, auth token expiration, internal errors
     # https://docs.microsoft.com/en-us/advertising/guides/services-protocol?view=bingads-13#throttling
     # https://docs.microsoft.com/en-us/advertising/guides/operation-error-codes?view=bingads-13
-    retry_on_codes: Iterator[int] = [117, 207, 4204, 109, 0]
+    retry_on_codes: Iterator[int] = ["117", "207", "4204", "109", "0"]
     max_retries: int = 3
     # A backoff factor to apply between attempts after the second try
     # {retry_factor} * (2 ** ({number of total retries} - 1))
     retry_factor: int = 15
+    # environments supported by Microsoft Advertising: sandbox, production
+    environment: str = "production"
+    # The time interval in milliseconds between two status polling attempts.
+    report_poll_interval: int = 15000
 
     def __init__(
         self,
@@ -55,6 +61,8 @@ def __init__(
         client_secret: str,
         client_id: str,
         refresh_token: str,
+        reports_start_date: str,
+        report_aggregation: Mapping[str, str],
         **kwargs: Mapping[str, Any],
     ) -> None:
         self.authorization_data: Mapping[str, AuthorizationData] = {}
@@ -67,8 +75,10 @@ def __init__(
         self.refresh_token = refresh_token
         self.customer_id = customer_id
         self.developer_token = developer_token
+        self.report_aggregation = report_aggregation["value"]
 
         self.oauth: OAuthTokens = self._get_access_token()
+        self.reports_start_date = pendulum.parse(reports_start_date).astimezone(tz=timezone.utc)
 
     @lru_cache(maxsize=None)
     def _get_auth_data(self, account_id: Optional[str] = None) -> AuthorizationData:
@@ -95,18 +105,19 @@ def is_token_expiring(self) -> bool:
         return False if token_updated_expires_in > self.refresh_token_safe_delta else True
 
     def should_retry(self, error: WebFault) -> bool:
-        error_code = errorcode_of_exception(error)
+        error_code = str(errorcode_of_exception(error))
         give_up = error_code not in self.retry_on_codes
         if give_up:
-            self.logger.info(f"Giving up for returned error code: {error_code}")
+            self.logger.error(f"Giving up for returned error code: {error_code}. Error details: {self._get_error_message(error)}")
         return give_up
 
+    def _get_error_message(self, error: WebFault) -> str:
+        return str(self.asdict(error.fault)) if hasattr(error, "fault") else str(error)
+
     def log_retry_attempt(self, details: Mapping[str, Any]) -> None:
         _, exc, _ = sys.exc_info()
-        error = self.asdict(exc.fault) if hasattr(exc, "fault") else exc
-
         self.logger.info(
-            f"Caught retryable error: {str(error)} after {details['tries']} tries. Waiting {details['wait']} seconds then retrying..."
+            f"Caught retryable error: {self._get_error_message(exc)} after {details['tries']} tries. Waiting {details['wait']} seconds then retrying..."
         )
 
     def request(self, **kwargs: Mapping[str, Any]) -> Mapping[str, Any]:
@@ -122,18 +133,23 @@ def request(self, **kwargs: Mapping[str, Any]) -> Mapping[str, Any]:
 
     def _request(
         self,
-        service_name: str,
+        service_name: Optional[str],
         operation_name: str,
         account_id: Optional[str],
         params: Mapping[str, Any],
+        is_report_service: bool = False,
     ) -> Mapping[str, Any]:
         """
         Executes appropriate Service Operation on Bing Ads API
         """
         if self.is_token_expiring():
             self.oauth = self._get_access_token()
 
-        service = self.get_service(service_name=service_name, account_id=account_id)
+        if is_report_service:
+            service = self._get_reporting_service(account_id=account_id)
+        else:
+            service = self.get_service(service_name=service_name, account_id=account_id)
+
         return getattr(service, operation_name)(**params)
 
     @lru_cache(maxsize=None)
@@ -146,8 +162,18 @@ def get_service(
             service=service_name,
             version=self.api_version,
             authorization_data=self._get_auth_data(account_id),
-            # environments supported by Microsoft Advertising: sandbox, production
-            environment="production",
+            environment=self.environment,
+        )
+
+    @lru_cache(maxsize=None)
+    def _get_reporting_service(
+        self,
+        account_id: Optional[str] = None,
+    ) -> ServiceClient:
+        return ReportingServiceManager(
+            authorization_data=self._get_auth_data(account_id),
+            poll_interval_in_milliseconds=self.report_poll_interval,
+            environment=self.environment,
         )
 
     @classmethod