Source bing ads fix add lb window

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.19
+LABEL io.airbyte.version=0.1.20
 LABEL io.airbyte.name=airbyte/source-bing-ads

airbyte-integrations/connectors/source-bing-ads/bootstrap.md

@@ -33,6 +33,8 @@ Connector uses `reports_start_date` config for initial reports sync and current
 
 Connector has `hourly_reports`, `daily_reports`, `weekly_reports`, `monthly_reports` report streams. For example `account_performance_report_daily`, `ad_group_performance_report_weekly`. All these reports streams will be generated on execute.
 
+If `lookback_window` is set to a non-null value, initial reports sync will start at `reports_start_date - lookback_window`. Following reports sync will start at `cursor_value - lookback_window`. 
+
 ## Request caching
 
 Based on [library](https://vcrpy.readthedocs.io/en/latest/)

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

@@ -4,6 +4,7 @@
   "client_id": "123",
   "developer_token": "asgag4gwag3",
   "reports_start_date": "2018-11-13",
+  "lookback_window": 0,
   "hourly_reports": false,
   "daily_reports": false,
   "weekly_reports": true,

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

@@ -203,11 +203,7 @@ def get_request_date(self, reporting_service: ServiceClient, date: datetime) ->
     def request_params(
         self, stream_state: Mapping[str, Any] = None, account_id: str = None, **kwargs: Mapping[str, Any]
     ) -> Mapping[str, Any]:
-        if not stream_state or not account_id or not stream_state.get(account_id, {}).get(self.cursor_field):
-            start_date = self.client.reports_start_date
-        else:
-            # gets starting point for a stream and account
-            start_date = pendulum.from_timestamp(stream_state[account_id][self.cursor_field])
+        start_date = self.get_start_date(stream_state, account_id)
 
         reporting_service = self.client.get_service("ReportingService")
         request_time_zone = reporting_service.factory.create("ReportTimeZone")
@@ -228,6 +224,13 @@ def request_params(
             "timeout_in_milliseconds": self.timeout,
         }
 
+    def get_start_date(self, stream_state: Mapping[str, Any] = None, account_id: str = None):
+        if stream_state and account_id:
+            if stream_state.get(account_id, {}).get(self.cursor_field):
+                return pendulum.from_timestamp(stream_state[account_id][self.cursor_field])
+
+        return self.client.reports_start_date
+
     def get_updated_state(
         self,
         current_stream_state: MutableMapping[str, Any],
@@ -337,3 +340,15 @@ def stream_slices(
             yield {"account_id": account["Id"], "customer_id": account["ParentCustomerId"]}
 
         yield from []
+
+
+class PerformanceReportsMixin(ReportsMixin):
+    def get_start_date(self, stream_state: Mapping[str, Any] = None, account_id: str = None):
+        start_date = super().get_start_date(stream_state, account_id)
+
+        if self.config.get("lookback_window"):
+            # Datetime subtract won't work with days = 0
+            # it'll output an AirbuteError
+            return start_date.subtract(days=self.config["lookback_window"])
+        else:
+            return start_date

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

@@ -23,6 +23,7 @@
     HISTORICAL_FIELDS,
     LOW_QUALITY_FIELDS,
     REVENUE_FIELDS,
+    PerformanceReportsMixin,
     ReportsMixin,
 )
 from suds import sudsobject
@@ -343,7 +344,7 @@ class BudgetSummaryReport(ReportsMixin, BingAdsStream):
     ]
 
 
-class CampaignPerformanceReport(ReportsMixin, BingAdsStream):
+class CampaignPerformanceReport(PerformanceReportsMixin, BingAdsStream):
     data_field: str = ""
     service_name: str = "ReportingService"
     report_name: str = "CampaignPerformanceReport"
@@ -427,7 +428,7 @@ class CampaignPerformanceReportMonthly(CampaignPerformanceReport):
     ]
 
 
-class AdPerformanceReport(ReportsMixin, BingAdsStream):
+class AdPerformanceReport(PerformanceReportsMixin, BingAdsStream):
     data_field: str = ""
     service_name: str = "ReportingService"
     report_name: str = "AdPerformanceReport"
@@ -499,7 +500,7 @@ class AdPerformanceReportMonthly(AdPerformanceReport):
     report_aggregation = "Monthly"
 
 
-class AdGroupPerformanceReport(ReportsMixin, BingAdsStream):
+class AdGroupPerformanceReport(PerformanceReportsMixin, BingAdsStream):
     data_field: str = ""
     service_name: str = "ReportingService"
     report_name: str = "AdGroupPerformanceReport"
@@ -586,7 +587,7 @@ class AdGroupPerformanceReportMonthly(AdGroupPerformanceReport):
     ]
 
 
-class KeywordPerformanceReport(ReportsMixin, BingAdsStream):
+class KeywordPerformanceReport(PerformanceReportsMixin, BingAdsStream):
     data_field: str = ""
     service_name: str = "ReportingService"
     report_name: str = "KeywordPerformanceReport"
@@ -671,7 +672,7 @@ class KeywordPerformanceReportMonthly(KeywordPerformanceReport):
     report_aggregation = "Monthly"
 
 
-class AccountPerformanceReport(ReportsMixin, BingAdsStream):
+class AccountPerformanceReport(PerformanceReportsMixin, BingAdsStream):
     data_field: str = ""
     service_name: str = "ReportingService"
     report_name: str = "AccountPerformanceReport"

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

@@ -60,6 +60,15 @@
         "default": "2020-01-01",
         "description": "The start date from which to begin replicating report data. Any data generated before this date will not be replicated in reports. This is a UTC date in YYYY-MM-DD format.",
         "order": 5
+      },
+      "lookback_window": {
+        "title": "Lookback window",
+        "description": "Also known as attribution or conversion window. How far into the past to look for records (in days). If your conversion window has an hours/minutes granularity, round it up to the number of days exceeding. Used only for performance report streams in incremental mode.",
+        "type": "integer",
+        "default": 0,
+        "minimum": 0,
+        "maximum": 90,
+        "order": 6
       }
     }
   },

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

@@ -6,7 +6,7 @@
 
 import pendulum
 from bingads.v13.internal.reporting.row_report_iterator import _RowReportRecord, _RowValues
-from source_bing_ads.reports import ReportsMixin
+from source_bing_ads.reports import PerformanceReportsMixin, ReportsMixin
 from source_bing_ads.source import SourceBingAds
 
 
@@ -23,6 +23,15 @@ def __init__(self) -> None:
         self.client = TestClient()
 
 
+class TestPerformanceReport(PerformanceReportsMixin, SourceBingAds):
+    date_format, report_columns, report_name, cursor_field = "YYYY-MM-DD", None, None, "Time"
+    report_aggregation = "Monthly"
+    report_schema_name = "campaign_performance_report"
+
+    def __init__(self) -> None:
+        self.client = TestClient()
+
+
 def test_get_column_value():
     row_values = _RowValues(
         {"AccountId": 1, "AverageCpc": 3, "AdGroupId": 2, "AccountName": 5, "Spend": 4},
@@ -38,6 +47,14 @@ def test_get_column_value():
     assert test_report.get_column_value(record, "Spend") == 1.203
 
 
+def test_get_updated_state_init_state():
+    test_report = TestReport()
+    stream_state = {}
+    latest_record = {"AccountId": 123, "Time": "2020-01-02"}
+    new_state = test_report.get_updated_state(stream_state, latest_record)
+    assert new_state["123"]["Time"] == (pendulum.parse("2020-01-02")).timestamp()
+
+
 def test_get_updated_state_new_state():
     test_report = TestReport()
     stream_state = {"123": {"Time": pendulum.parse("2020-01-01").timestamp()}}
@@ -79,3 +96,44 @@ def test_get_report_record_timestamp_hourly():
     test_report = TestReport()
     test_report.report_aggregation = "Hourly"
     assert pendulum.parse("2020-01-01T15:00:00").timestamp() == test_report.get_report_record_timestamp("2020-01-01|15")
+
+
+def test_report_get_start_date_wo_stream_state():
+    expected_start_date = "2020-01-01"
+    test_report = TestReport()
+    test_report.client.reports_start_date = "2020-01-01"
+    stream_state = {}
+    account_id = "123"
+    assert expected_start_date == test_report.get_start_date(stream_state, account_id)
+
+
+def test_report_get_start_date_with_stream_state():
+    expected_start_date = pendulum.parse("2023-04-17T21:29:57")
+    test_report = TestReport()
+    test_report.cursor_field = "cursor_field"
+    test_report.client.reports_start_date = "2020-01-01"
+    stream_state = {"123": {"cursor_field": 1681766997}}
+    account_id = "123"
+    assert expected_start_date == test_report.get_start_date(stream_state, account_id)
+
+
+def test_report_get_start_date_performance_report_with_stream_state():
+    expected_start_date = pendulum.parse("2023-04-07T21:29:57")
+    test_report = TestPerformanceReport()
+    test_report.cursor_field = "cursor_field"
+    test_report.config = {"lookback_window": 10}
+    stream_state = {"123": {"cursor_field": 1681766997}}
+    account_id = "123"
+    assert expected_start_date == test_report.get_start_date(stream_state, account_id)
+
+
+def test_report_get_start_date_performance_report_wo_stream_state():
+    days_to_subtract = 10
+    reports_start_date = pendulum.parse("2021-04-07T00:00:00")
+    test_report = TestPerformanceReport()
+    test_report.cursor_field = "cursor_field"
+    test_report.client.reports_start_date = reports_start_date
+    test_report.config = {"lookback_window": days_to_subtract}
+    stream_state = {}
+    account_id = "123"
+    assert reports_start_date.subtract(days=days_to_subtract) == test_report.get_start_date(stream_state, account_id)