Source Facebook Marketing: Add lookback window to insights streams

airbyte-integrations/connectors/source-facebook-marketing/Dockerfile

@@ -13,5 +13,5 @@ ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
 
-LABEL io.airbyte.version=0.2.47
+LABEL io.airbyte.version=0.2.48
 LABEL io.airbyte.name=airbyte/source-facebook-marketing

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/source.py

@@ -64,6 +64,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
             api=api,
             start_date=config.start_date,
             end_date=config.end_date,
+            insights_lookback_window=config.insights_lookback_window
         )
         streams = [
             AdAccount(api=api),
@@ -159,6 +160,7 @@ def _update_insights_streams(self, insights: List[InsightConfig], default_args,
                 time_increment=insight.time_increment,
                 start_date=insight.start_date or default_args["start_date"],
                 end_date=insight.end_date or default_args["end_date"],
+                insights_lookback_window=insight.insights_lookback_window or default_args["insights_lookback_window"],
             )
             insight_stream = AdsInsights(**args)
             insights_custom_streams.append(insight_stream)

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/spec.py

@@ -77,6 +77,14 @@ class Config:
         pattern=DATE_TIME_PATTERN,
         examples=["2017-01-26T00:00:00Z"],
     )
+    insights_lookback_window: Optional[PositiveInt] = Field(
+        title="Custom Insights Lookback Window",
+        description=(
+            "The attribution window"
+        ),
+        exclusiveMaximum=28,
+        default=28,
+    )
 
 
 class ConnectorConfig(BaseConfig):
@@ -156,3 +164,13 @@ class Config:
             "Page size used when sending requests to Facebook API to specify number of records per page when response has pagination. Most users do not need to set this field unless they specifically need to tune the connector to address specific issues or use cases."
         ),
     )
+
+    insights_lookback_window: Optional[PositiveInt] = Field(
+        title="Insights Lookback Window",
+        order=7,
+        description=(
+            "The attribution window"
+        ),
+        exclusiveMaximum=28,
+        default=28,
+    )

airbyte-integrations/connectors/source-facebook-marketing/source_facebook_marketing/streams/base_insight_streams.py

@@ -46,9 +46,6 @@ class AdsInsights(FBMarketingIncrementalStream):
     # HTTP response.
     # https://developers.facebook.com/docs/marketing-api/reference/ad-account/insights/#overview
     INSIGHTS_RETENTION_PERIOD = pendulum.duration(months=37)
-    # Facebook freezes insight data 28 days after it was generated, which means that all data
-    # from the past 28 days may have changed since we last emitted it, so we retrieve it again.
-    INSIGHTS_LOOKBACK_PERIOD = pendulum.duration(days=28)
 
     action_breakdowns = ALL_ACTION_BREAKDOWNS
     level = "ad"
@@ -64,6 +61,7 @@ def __init__(
         breakdowns: List[str] = None,
         action_breakdowns: List[str] = None,
         time_increment: Optional[int] = None,
+        insights_lookback_window: int = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -74,6 +72,7 @@ def __init__(
         self.breakdowns = breakdowns or self.breakdowns
         self.time_increment = time_increment or self.time_increment
         self._new_class_name = name
+        self._insights_lookback_window = insights_lookback_window
 
         # state
         self._cursor_value: Optional[pendulum.Date] = None  # latest period that was read
@@ -91,6 +90,16 @@ def primary_key(self) -> Optional[Union[str, List[str], List[List[str]]]]:
         """Build complex PK based on slices and breakdowns"""
         return ["date_start", "account_id", "ad_id"] + self.breakdowns
 
+    @property
+    def insights_lookback_period(self):
+        """
+        Facebook freezes insight data 28 days after it was generated, which means that all data
+        from the past 28 days may have changed since we last emitted it, so we retrieve it again.
+        But in some cases users my have define their own lookback window, thats
+        why the value for `insights_lookback_window` is set throught config.
+        """
+        return pendulum.duration(days=self._insights_lookback_window)
+
     def list_objects(self, params: Mapping[str, Any]) -> Iterable:
         """Because insights has very different read_records we don't need this method anymore"""
 
@@ -217,13 +226,12 @@ def _get_start_date(self) -> pendulum.Date:
         """
         today = pendulum.today().date()
         oldest_date = today - self.INSIGHTS_RETENTION_PERIOD
-        refresh_date = today - self.INSIGHTS_LOOKBACK_PERIOD
-
+        refresh_date = today - self.insights_lookback_period
         if self._cursor_value:
             start_date = self._cursor_value + pendulum.duration(days=self.time_increment)
             if start_date > refresh_date:
                 logger.info(
-                    f"The cursor value within refresh period ({self.INSIGHTS_LOOKBACK_PERIOD}), start sync from {refresh_date} instead."
+                    f"The cursor value within refresh period ({self.insights_lookback_period}), start sync from {refresh_date} instead."
                 )
             start_date = min(start_date, refresh_date)
 
@@ -234,7 +242,6 @@ def _get_start_date(self) -> pendulum.Date:
             start_date = self._start_date
         if start_date < oldest_date:
             logger.warning(f"Loading insights older then {self.INSIGHTS_RETENTION_PERIOD} is not possible. Start sync from {oldest_date}.")
-
         return max(oldest_date, start_date)
 
     def request_params(self, **kwargs) -> MutableMapping[str, Any]:

docs/integrations/sources/facebook-marketing.md

@@ -108,6 +108,8 @@ For more information, see the [Facebook Insights API documentation.](https://dev
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                                                                                                                                                                                                   |
 |:--------|:-----------|:---------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.2.48  | 2022-04-27 | [12402](https://github.com/airbytehq/airbyte/pull/12402) |  Add lookback window to insgiths streams                                                     |
+
 | 0.2.47  | 2022-05-06 | [12685](https://github.com/airbytehq/airbyte/pull/12685) | Update CDK to v0.1.56 to emit an `AirbyeTraceMessage` on uncaught exceptions                                                                                                                                                                                                                              |
 | 0.2.46  | 2022-04-22 | [12171](https://github.com/airbytehq/airbyte/pull/12171) | Allow configuration of page_size for requests                                                                                                                                                                                                                                                             |
 | 0.2.45  | 2022-05-03 | [12390](https://github.com/airbytehq/airbyte/pull/12390) | Better retry logic for split-up async jobs                                                                                                                                                                                                                                                                |