✨ Source Bing Ads: custom reports (#32306)

Co-authored-by: darynaishchenko <darynaishchenko@users.noreply.github.com>

Author: darynaishchenko

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.12.1
+ dockerImageTag: 1.13.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,
@@ -83,17 +85,49 @@ 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:
+ is_valid, reason = custom_report.validate_report_configuration()
+ if not is_valid:
+ raise AirbyteTracedException(
+ message=f"Config validation error: {custom_report.name}: {reason}",
+ internal_message=f"{custom_report.name}: {reason}",
+ failure_type=FailureType.config_error,
+ )
+
+ def _clear_reporting_object_name(self, report_object: str) -> str:
+ # reporting mixin adds 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._clear_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 +161,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

@@ -63,6 +63,105 @@
  "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",
+ "enum": [
+ "AccountPerformanceReportRequest",
+ "AdDynamicTextPerformanceReportRequest",
+ "AdExtensionByAdReportRequest",
+ "AdExtensionByKeywordReportRequest",
+ "AdExtensionDetailReportRequest",
+ "AdGroupPerformanceReportRequest",
+ "AdPerformanceReportRequest",
+ "AgeGenderAudienceReportRequest",
+ "AudiencePerformanceReportRequest",
+ "CallDetailReportRequest",
+ "CampaignPerformanceReportRequest",
+ "ConversionPerformanceReportRequest",
+ "DestinationUrlPerformanceReportRequest",
+ "DSAAutoTargetPerformanceReportRequest",
+ "DSACategoryPerformanceReportRequest",
+ "DSASearchQueryPerformanceReportRequest",
+ "GeographicPerformanceReportRequest",
+ "GoalsAndFunnelsReportRequest",
+ "HotelDimensionPerformanceReportRequest",
+ "HotelGroupPerformanceReportRequest",
+ "KeywordPerformanceReportRequest",
+ "NegativeKeywordConflictReportRequest",
+ "ProductDimensionPerformanceReportRequest",
+ "ProductMatchCountReportRequest",
+ "ProductNegativeKeywordConflictReportRequest",
+ "ProductPartitionPerformanceReportRequest",
+ "ProductPartitionUnitPerformanceReportRequest",
+ "ProductSearchQueryPerformanceReportRequest",
+ "ProfessionalDemographicsAudienceReportRequest",
+ "PublisherUsagePerformanceReportRequest",
+ "SearchCampaignChangeHistoryReportRequest",
+ "SearchQueryPerformanceReportRequest",
+ "ShareOfVoiceReportRequest",
+ "UserLocationPerformanceReportRequest"
+ ]
+ },
+ "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",
+ "DayOfWeek",
+ "HourOfDay",
+ "WeeklyStartingMonday",
+ "Summary"
+ ]
+ },
+ "default": ["Hourly"]
+ }
+ },
+ "required": [
+ "name",
+ "reporting_object",
+ "report_columns",
+ "report_aggregation"
+ ]
+ }
  }
  }
  },

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

@@ -2,12 +2,15 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 import os
+import re
 import ssl
 import time
+import xml.etree.ElementTree as ET
 from abc import ABC, abstractmethod
 from datetime import timezone
 from typing import Any, Iterable, List, Mapping, MutableMapping, Optional, Tuple, Union
 from urllib.error import URLError
+from urllib.parse import urlparse
 
 import _csv
 import pandas as pd
@@ -16,6 +19,7 @@
 from airbyte_cdk.sources.streams import IncrementalMixin, Stream
 from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
 from bingads.service_client import ServiceClient
+from bingads.v13.internal.reporting.row_report import _RowReport
 from bingads.v13.internal.reporting.row_report_iterator import _RowReportRecord
 from bingads.v13.reporting.reporting_service_manager import ReportingServiceManager
 from numpy import nan
@@ -32,7 +36,7 @@
  PerformanceReportsMixin,
  ReportsMixin,
 )
-from suds import sudsobject
+from suds import WebFault, sudsobject
 
 
 class BingAdsBaseStream(Stream, ABC):
@@ -1260,3 +1264,95 @@ class UserLocationPerformanceReportWeekly(UserLocationPerformanceReport):
 
 class UserLocationPerformanceReportMonthly(UserLocationPerformanceReport):
  report_aggregation = "Monthly"
+
+
+class CustomReport(PerformanceReportsMixin, BingAdsReportingServiceStream, ABC):
+ transformer: TypeTransformer = TypeTransformer(TransformConfig.DefaultSchemaNormalization)
+ custom_report_columns = []
+ report_schema_name = None
+ primary_key = None
+
+ @property
+ def cursor_field(self) -> Union[str, List[str]]:
+ # Summary aggregation doesn't include TimePeriod field
+ if self.report_aggregation != "Summary":
+ return "TimePeriod"
+
+ @property
+ def report_columns(self):
+ # adding common and default columns
+ if "AccountId" not in self.custom_report_columns:
+ self.custom_report_columns.append("AccountId")
+ if self.cursor_field and self.cursor_field not in self.custom_report_columns:
+ self.custom_report_columns.append(self.cursor_field)
+ 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
+
+ def validate_report_configuration(self) -> Tuple[bool, str]:
+ # gets /bingads/v13/proxies/production/reporting_service.xml
+ reporting_service_file = self.client.get_service(self.service_name)._get_service_info_dict(self.client.api_version)[
+ ("reporting", self.client.environment)
+ ]
+ tree = ET.parse(urlparse(reporting_service_file).path)
+ request_object = tree.find(f".//{{*}}complexType[@name='{self.report_name}Request']")
+
+ report_object_columns = self._get_object_columns(request_object, tree)
+ is_custom_cols_in_report_object_cols = all(x in report_object_columns for x in self.custom_report_columns)
+
+ if not is_custom_cols_in_report_object_cols:
+ return False, (
+ f"Reporting Columns are invalid. Columns that you provided don't belong to Reporting Data Object Columns:"
+ f" {self.custom_report_columns}. Please ensure it is correct in Bing Ads Docs."
+ )
+
+ return True, ""
+
+ def _clear_namespace(self, type: str) -> str:
+ return re.sub(r"^[a-z]+:", "", type)
+
+ def _get_object_columns(self, request_el: ET.Element, tree: ET.ElementTree) -> List[str]:
+ column_el = request_el.find(".//{*}element[@name='Columns']")
+ array_of_columns_name = self._clear_namespace(column_el.get("type"))
+
+ array_of_columns_elements = tree.find(f".//{{*}}complexType[@name='{array_of_columns_name}']")
+ inner_array_of_columns_elements = array_of_columns_elements.find(".//{*}element")
+ column_el_name = self._clear_namespace(inner_array_of_columns_elements.get("type"))
+
+ column_el = tree.find(f".//{{*}}simpleType[@name='{column_el_name}']")
+ column_enum_items = column_el.findall(".//{*}enumeration")
+ column_enum_items_values = [el.get("value") for el in column_enum_items]
+ return column_enum_items_values
+
+ def get_report_record_timestamp(self, datestring: str) -> int:
+ """
+ Parse report date field based on aggregation type
+ """
+ if not self.report_aggregation:
+ date = pendulum.from_format(datestring, "M/D/YYYY")
+ else:
+ if self.report_aggregation in ["DayOfWeek", "HourOfDay"]:
+ return int(datestring)
+ if self.report_aggregation == "Hourly":
+ date = pendulum.from_format(datestring, "YYYY-MM-DD|H")
+ else:
+ date = pendulum.parse(datestring)
+
+ return date.int_timestamp
+
+ def send_request(self, params: Mapping[str, Any], customer_id: str, account_id: str) -> _RowReport:
+ try:
+ return super().send_request(params, customer_id, account_id)
+ except WebFault as e:
+ self.logger.error(
+ f"Could not sync custom report {self.name}: Please validate your column and aggregation configuration. "
+ f"Error form server: [{e.fault.faultstring}]"
+ )

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

@@ -12,6 +12,7 @@
 import source_bing_ads.client
 from airbyte_cdk.utils import AirbyteTracedException
 from bingads.authorization import AuthorizationData, OAuthTokens
+from bingads.v13.bulk import BulkServiceManager
 from bingads.v13.reporting.exceptions import ReportingDownloadException
 from suds import sudsobject
 
@@ -176,3 +177,15 @@ def test_bulk_service_manager(patched_request_tokens):
  client = source_bing_ads.client.Client("tenant_id", "2020-01-01", client_id="client_id", refresh_token="refresh_token")
  service = client._bulk_service_manager()
  assert (service._poll_interval_in_milliseconds, service._environment) == (5000, client.environment)
+
+
+def test_get_bulk_entity(requests_mock):
+ requests_mock.post(
+ "https://login.microsoftonline.com/tenant_id/oauth2/v2.0/token",
+ status_code=200,
+ json={"access_token": "test", "expires_in": "9000", "refresh_token": "test"},
+ )
+ client = source_bing_ads.client.Client("tenant_id", "2020-01-01", client_id="client_id", refresh_token="refresh_token")
+ with patch.object(BulkServiceManager, "download_file", return_value="file.csv"):
+ bulk_entity = client.get_bulk_entity(data_scope=["EntityData"], download_entities=["AppInstallAds"])
+ assert bulk_entity == "file.csv"