✨ Source Bing Ads: custom reports

airbyte-integrations/connectors/source-bing-ads/metadata.yaml

@@ -16,7 +16,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: 47f25999-dd5e-4636-8c39-e7cea2453331
-  dockerImageTag: 1.11.0
+  dockerImageTag: 1.12.0
   dockerRepository: airbyte/source-bing-ads
   documentationUrl: https://docs.airbyte.com/integrations/sources/bing-ads
   githubIssueLabel: source-bing-ads

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

@@ -37,7 +37,7 @@ class Client:
     # 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[str] = ["117", "207", "4204", "109", "0"]
-    max_retries: int = 10
+    max_retries: int = 5
     # A backoff factor to apply between attempts after the second try
     # {retry_factor} * (2 ** ({number of total retries} - 1))
     retry_factor: int = 15

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

@@ -2,7 +2,7 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 from itertools import product
-from typing import Any, List, Mapping, Tuple
+from typing import Any, List, Mapping, Optional, Tuple
 
 from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.models import FailureType, SyncMode
@@ -41,6 +41,7 @@
     AgeGenderAudienceReportWeekly,
     AppInstallAdLabels,
     AppInstallAds,
+    BingAdsReportingServiceStream,
     BudgetSummaryReport,
     CampaignImpressionPerformanceReportDaily,
     CampaignImpressionPerformanceReportHourly,
@@ -52,6 +53,7 @@
     CampaignPerformanceReportMonthly,
     CampaignPerformanceReportWeekly,
     Campaigns,
+    CustomReport,
     GeographicPerformanceReportDaily,
     GeographicPerformanceReportHourly,
     GeographicPerformanceReportMonthly,
@@ -72,6 +74,7 @@
     UserLocationPerformanceReportMonthly,
     UserLocationPerformanceReportWeekly,
 )
+from suds import TypeNotFound, WebFault
 
 
 class SourceBingAds(AbstractSource):
@@ -83,17 +86,66 @@ def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) ->
         try:
             client = Client(**config)
             account_ids = {str(account["Id"]) for account in Accounts(client, config).read_records(SyncMode.full_refresh)}
+            self.validate_custom_reposts(config, client)
             if account_ids:
                 return True, None
             else:
                 raise AirbyteTracedException(
-                    message="Config validation error: You don't have accounts assigned to this user.",
+                    message="Config validation error: You don't have accounts assigned to this user. Please verify your developer token.",
                     internal_message="You don't have accounts assigned to this user.",
                     failure_type=FailureType.config_error,
                 )
         except Exception as error:
             return False, error
 
+    def validate_custom_reposts(self, config: Mapping[str, Any], client: Client):
+        custom_reports = self.get_custom_reports(config, client)
+        for custom_report in custom_reports:
+            try:
+                for account in Accounts(client, config).read_records(SyncMode.full_refresh):
+                    list(
+                        custom_report.read_records(
+                            sync_mode=SyncMode.full_refresh,
+                            stream_slice={"account_id": account["Id"], "customer_id": account["ParentCustomerId"]},
+                        )
+                    )
+            except TypeNotFound:
+                raise AirbyteTracedException(
+                    message=f"Config validation error: You have provided invalid Reporting Object: {custom_report.report_name}. "
+                    f"Please verify it in Bing Ads Docs"
+                    f" https://learn.microsoft.com/en-us/advertising/reporting-service/reporting-service-reference?view=bingads-13",
+                    internal_message="invalid reporting object was provided.",
+                    failure_type=FailureType.config_error,
+                )
+            except WebFault as e:
+                raise AirbyteTracedException(
+                    message=f"Config validation error: You have provided invalid Reporting Columns: {custom_report.custom_report_columns}. "
+                    f"Make sure that you provided right columns for this report, not all columns can be added/removed."
+                    f"Please, verify it",
+                    internal_message=f"invalid reporting columns were provided. {e}",
+                    failure_type=FailureType.config_error,
+                )
+
+    def _validate_reporting_object_name(self, report_object: str) -> str:
+        # reporting mixin adds it if user didn't provide it
+        if report_object.endswith("Request"):
+            return report_object.replace("Request", "")
+        return report_object
+
+    def get_custom_reports(self, config: Mapping[str, Any], client: Client) -> List[Optional[Stream]]:
+        return [
+            type(
+                report["name"],
+                (CustomReport,),
+                {
+                    "report_name": self._validate_reporting_object_name(report["reporting_object"]),
+                    "custom_report_columns": report["report_columns"],
+                    "report_aggregation": report["report_aggregation"],
+                },
+            )(client, config)
+            for report in config.get("custom_reports", [])
+        ]
+
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         client = Client(**config)
         streams = [
@@ -127,4 +179,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         )
         report_aggregation = ("Hourly", "Daily", "Weekly", "Monthly")
         streams.extend([eval(f"{report}{aggregation}")(client, config) for (report, aggregation) in product(reports, report_aggregation)])
+
+        custom_reports = self.get_custom_reports(config, client)
+        streams.extend(custom_reports)
         return streams

airbyte-integrations/connectors/source-bing-ads/source_bing_ads/spec.json

@@ -69,6 +69,64 @@
         "minimum": 0,
         "maximum": 90,
         "order": 6
+      },
+      "custom_reports": {
+        "title": "Custom Reports",
+        "description": "You can add your Custom Bing Ads report by creating one.",
+        "order": 7,
+        "type": "array",
+        "items": {
+          "title": "Custom Report Config",
+          "type": "object",
+          "properties": {
+            "name": {
+              "title": "Report Name",
+              "description": "The name of the custom report, this name would be used as stream name",
+              "type": "string",
+              "examples": [
+                "Account Performance",
+                "AdDynamicTextPerformanceReport",
+                "custom report"
+              ]
+            },
+            "reporting_object": {
+              "title": "Reporting Data Object",
+              "description": "The name of the the object derives from the ReportRequest object. You can find it in Bing Ads Api docs - Reporting API - Reporting Data Objects.",
+              "type": "string",
+              "examples": [
+                "AccountPerformanceReport",
+                "AdDynamicTextPerformanceReport"
+              ]
+            },
+            "report_columns": {
+              "title": "Columns",
+              "description": "A list of available report object columns. You can find it in description of reporting object that you want to add to custom report.",
+              "type": "array",
+              "items": {
+                "description": "Name of report column.",
+                "type": "string"
+              },
+              "minItems": 1
+            },
+            "report_aggregation": {
+              "title": "Aggregation",
+              "description": "A list of available aggregations.",
+              "type": "string",
+              "items": {
+                "title": "ValidEnums",
+                "description": "An enumeration of aggregations.",
+                "enum": ["Hourly", "Daily", "Weekly", "Monthly"]
+              },
+              "default": ["Hourly"]
+            }
+          },
+          "required": [
+            "name",
+            "reporting_object",
+            "report_columns",
+            "report_aggregation"
+          ]
+        }
       }
     }
   },

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

@@ -1260,3 +1260,28 @@ class UserLocationPerformanceReportWeekly(UserLocationPerformanceReport):
 
 class UserLocationPerformanceReportMonthly(UserLocationPerformanceReport):
     report_aggregation = "Monthly"
+
+
+class CustomReport(PerformanceReportsMixin, BingAdsReportingServiceStream, ABC):
+    transformer: TypeTransformer = TypeTransformer(TransformConfig.DefaultSchemaNormalization)
+    custom_report_columns = []
+    cursor_field = "TimePeriod"
+    report_schema_name = None
+    primary_key = None
+
+    @property
+    def report_columns(self):
+        # adding common and default columns
+        if "AccountId" not in self.custom_report_columns:
+            self.custom_report_columns.append("AccountId")
+        return list(frozenset(self.custom_report_columns))
+
+    def get_json_schema(self) -> Mapping[str, Any]:
+        columns_schema = {col: {"type": ["null", "string"]} for col in self.report_columns}
+        schema: Mapping[str, Any] = {
+            "$schema": "https://json-schema.org/draft-07/schema#",
+            "type": ["null", "object"],
+            "additionalProperties": True,
+            "properties": columns_schema,
+        }
+        return schema

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

@@ -8,8 +8,11 @@
 import pytest
 import source_bing_ads
 from airbyte_cdk.models import SyncMode
+from airbyte_cdk.utils import AirbyteTracedException
+from source_bing_ads.reports import ReportsMixin
 from source_bing_ads.source import SourceBingAds
-from source_bing_ads.streams import AccountPerformanceReportMonthly, Accounts, AdGroups, Ads, AppInstallAds, Campaigns
+from source_bing_ads.streams import AccountPerformanceReportMonthly, Accounts, AdGroups, Ads, AppInstallAds, BingAdsStream, Campaigns
+from suds import TypeNotFound, WebFault
 
 
 @pytest.fixture(name="config")
@@ -25,6 +28,44 @@ def config_fixture():
     }
 
 
+@pytest.fixture(name="config_with_custom_reports")
+def config_with_custom_reports_fixture():
+    """Generates streams settings with custom reports from a config file"""
+    return {
+        "tenant_id": "common",
+        "developer_token": "fake_developer_token",
+        "refresh_token": "fake_refresh_token",
+        "client_id": "fake_client_id",
+        "reports_start_date": "2020-01-01",
+        "lookback_window": 0,
+        "custom_reports": [
+            {
+                "name": "my test custom report",
+                "reporting_object": "DSAAutoTargetPerformanceReport",
+                "report_columns": [
+                    "AbsoluteTopImpressionRatePercent",
+                    "AccountId",
+                    "AccountName",
+                    "AccountNumber",
+                    "AccountStatus",
+                    "AdDistribution",
+                    "AdGroupId",
+                    "AdGroupName",
+                    "AdGroupStatus",
+                    "AdId",
+                    "AllConversionRate",
+                    "AllConversions",
+                    "AllConversionsQualified",
+                    "AllCostPerConversion",
+                    "AllReturnOnAdSpend",
+                    "AllRevenue",
+                ],
+                "report_aggregation": "Weekly",
+            }
+        ],
+    }
+
+
 @pytest.fixture(name="logger_mock")
 def logger_mock_fixture():
     return patch("source_bing_ads.source.AirbyteLogger")
@@ -47,7 +88,9 @@ def test_source_check_connection_failed_user_do_not_have_accounts(mocked_client,
     with patch.object(Accounts, "read_records", return_value=[]):
         connected, reason = SourceBingAds().check_connection(logger_mock, config=config)
         assert connected is False
-        assert reason.message == "Config validation error: You don't have accounts assigned to this user."
+        assert (
+            reason.message == "Config validation error: You don't have accounts assigned to this user. Please verify your developer token."
+        )
 
 
 def test_source_check_connection_failed_invalid_creds(config, logger_mock):
@@ -60,6 +103,69 @@ def test_source_check_connection_failed_invalid_creds(config, logger_mock):
         )
 
 
+@patch.object(source_bing_ads.source, "Client")
+def test_validate_custom_reposts(mocked_client, config_with_custom_reports, logger_mock):
+    with (
+        patch.object(
+            Accounts,
+            "read_records",
+            return_value=iter([{"Id": 180519267, "ParentCustomerId": 78798732}, {"Id": 180278106, "ParentCustomerId": 82372972}]),
+        )
+    ):
+        assert SourceBingAds().validate_custom_reposts(config=config_with_custom_reports, client=mocked_client) is None
+
+
+@patch.object(source_bing_ads.source, "Client")
+def test_validate_custom_reposts_failed_invalid_report_object(mocked_client, config_with_custom_reports, logger_mock):
+    with patch.object(
+        Accounts,
+        "read_records",
+        return_value=iter([{"Id": 180519267, "ParentCustomerId": 78798732}, {"Id": 180278106, "ParentCustomerId": 82372972}]),
+    ):
+        with patch.object(ReportsMixin, "get_report_request", side_effect=TypeNotFound(name="NonExistingReportingObject")):
+            with pytest.raises(AirbyteTracedException) as e:
+                SourceBingAds().validate_custom_reposts(config=config_with_custom_reports, client=mocked_client)
+
+    assert e.value.internal_message == "invalid reporting object was provided."
+    assert (
+        "Config validation error: You have provided invalid Reporting Object: DSAAutoTargetPerformanceReport. "
+        "Please verify it in Bing Ads Docs"
+    ) in e.value.message
+
+
+@patch.object(source_bing_ads.source, "Client")
+def test_validate_custom_reposts_failed_invalid_report_columns(mocked_client, config_with_custom_reports, logger_mock):
+    with patch.object(
+        Accounts,
+        "read_records",
+        return_value=iter([{"Id": 180519267, "ParentCustomerId": 78798732}, {"Id": 180278106, "ParentCustomerId": 82372972}]),
+    ):
+        with patch.object(
+            BingAdsStream, "read_records", side_effect=WebFault(fault="Invalid client data.", document="Invalid cleint data.")
+        ):
+            with pytest.raises(AirbyteTracedException) as e:
+                SourceBingAds().validate_custom_reposts(config=config_with_custom_reports, client=mocked_client)
+
+    assert e.value.internal_message == "invalid reporting columns were provided. "
+    assert ("Config validation error: You have provided invalid Reporting Columns: ['AbsoluteTopImpressionRatePercent'") in e.value.message
+
+
+@patch.object(source_bing_ads.source, "Client")
+def test_get_custom_reports(mocked_client, config_with_custom_reports):
+    custom_reports = SourceBingAds().get_custom_reports(config_with_custom_reports, mocked_client)
+    assert isinstance(custom_reports, list)
+    assert custom_reports[0].report_name == "DSAAutoTargetPerformanceReport"
+    assert custom_reports[0].report_aggregation == "Weekly"
+    assert "AccountId" in custom_reports[0].custom_report_columns
+
+
+def test_validate_reporting_object_name():
+    reporting_object = SourceBingAds()._validate_reporting_object_name("DSAAutoTargetPerformanceReportRequest")
+    assert reporting_object == "DSAAutoTargetPerformanceReport"
+    reporting_object = SourceBingAds()._validate_reporting_object_name("DSAAutoTargetPerformanceReport")
+    assert reporting_object == "DSAAutoTargetPerformanceReport"
+
+
 @patch.object(source_bing_ads.source, "Client")
 def test_campaigns_request_params(mocked_client, config):