Source Facebook Marketing: retry FacebookBadObjectError (#15788)

* #404 oncall: source facebook marketing - retry 'facebook bad object error'

* source facebook marketing: upd changelog

* auto-bump connector version [ci skip]

Co-authored-by: Octavia Squidington III <[email protected]>

Author: davydov-d

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

@@ -256,7 +256,7 @@
 - name: Facebook Marketing
   sourceDefinitionId: e7778cfc-e97c-4458-9ecb-b4f2bba8946c
   dockerRepository: airbyte/source-facebook-marketing
-  dockerImageTag: 0.2.59
+  dockerImageTag: 0.2.60
   documentationUrl: https://docs.airbyte.io/integrations/sources/facebook-marketing
   icon: facebook.svg
   sourceType: api

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

@@ -1857,7 +1857,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-facebook-marketing:0.2.59"
+- dockerImage: "airbyte/source-facebook-marketing:0.2.60"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/facebook-marketing"
     changelogUrl: "https://docs.airbyte.io/integrations/sources/facebook-marketing"

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.59
+LABEL io.airbyte.version=0.2.60
 LABEL io.airbyte.name=airbyte/source-facebook-marketing

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

@@ -8,18 +8,23 @@
 from enum import Enum
 from typing import Any, Iterator, List, Mapping, Optional, Type, Union
 
+import backoff
 import pendulum
 from facebook_business.adobjects.ad import Ad
 from facebook_business.adobjects.adaccount import AdAccount
 from facebook_business.adobjects.adreportrun import AdReportRun
 from facebook_business.adobjects.adset import AdSet
 from facebook_business.adobjects.campaign import Campaign
 from facebook_business.adobjects.objectparser import ObjectParser
-from facebook_business.api import FacebookAdsApi, FacebookAdsApiBatch, FacebookResponse
+from facebook_business.api import FacebookAdsApi, FacebookAdsApiBatch, FacebookBadObjectError, FacebookResponse
+from source_facebook_marketing.streams.common import retry_pattern
 
 logger = logging.getLogger("airbyte")
 
 
+backoff_policy = retry_pattern(backoff.expo, FacebookBadObjectError, max_tries=5, factor=5)
+
+
 def update_in_batch(api: FacebookAdsApi, jobs: List["AsyncJob"]):
     """Update status of each job in the list in a batch, making it most efficient way to update status.
 
@@ -338,6 +343,7 @@ def _check_status(self) -> bool:
 
         return False
 
+    @backoff_policy
     def get_result(self) -> Any:
         """Retrieve result of the finished job."""
         if not self._job or self.failed:

airbyte-integrations/connectors/source-facebook-marketing/unit_tests/test_async_job.py

@@ -14,7 +14,7 @@
 from facebook_business.adobjects.adset import AdSet
 from facebook_business.adobjects.adsinsights import AdsInsights
 from facebook_business.adobjects.campaign import Campaign
-from facebook_business.api import FacebookAdsApiBatch
+from facebook_business.api import FacebookAdsApiBatch, FacebookBadObjectError
 from source_facebook_marketing.api import MyFacebookAdsApi
 from source_facebook_marketing.streams.async_job import InsightAsyncJob, ParentAsyncJob, Status, update_in_batch
 
@@ -299,6 +299,18 @@ def test_get_result(self, job, adreport, api):
         assert result[0].export_all_data() == {"some_data": 123}
         assert result[1].export_all_data() == {"some_data": 77}
 
+    def test_get_result_retried(self, mocker, job, api):
+        job.start()
+        api.call().json.return_value = {"data": [{"some_data": 123}, {"some_data": 77}]}
+        ads_insights = AdsInsights(api=api)
+        ads_insights._set_data({"items": [{"some_data": 123}, {"some_data": 77}]})
+        with mocker.patch(
+            "facebook_business.adobjects.objectparser.ObjectParser.parse_multiple",
+            side_effect=[FacebookBadObjectError("Bad data to set object data"), ads_insights],
+        ):
+            # in case this is not retried, an error will be raised
+            job.get_result()
+
     def test_get_result_when_job_is_not_started(self, job):
         with pytest.raises(RuntimeError, match=r"Incorrect usage of get_result - the job is not started or failed"):
             job.get_result()

docs/integrations/sources/facebook-marketing.md

@@ -120,7 +120,8 @@ Please be informed that the connector uses the `lookback_window` parameter to pe
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                                                                                                                                                                                           |
 |:--------|:-----------|:---------------------------------------------------------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| 0.2.59  | 2022-08-04 | [15327](https://github.com/airbytehq/airbyte/pull/15327) | Shift date validation from config validation to stream method                                                                                                                                                                                                                                   |     
+| 0.2.60  | 2022-08-19 | [15788](https://github.com/airbytehq/airbyte/pull/15788) | Retry FacebookBadObjectError                                                                                                                                                                                                                                                                      |
+| 0.2.59  | 2022-08-04 | [15327](https://github.com/airbytehq/airbyte/pull/15327) | Shift date validation from config validation to stream method                                                                                                                                                                                                                                     |     
 | 0.2.58  | 2022-07-25 | [15012](https://github.com/airbytehq/airbyte/pull/15012) | Add `DATA_RETENTION_PERIOD`validation and fix `failed_delivery_checks` field schema type issue                                                                                                                                                                                                    |     
 | 0.2.57  | 2022-07-25 | [14831](https://github.com/airbytehq/airbyte/pull/14831) | Update Facebook SDK to version 14.0.0                                                                                                                                                                                                                                                             | 
 | 0.2.56  | 2022-07-19 | [14831](https://github.com/airbytehq/airbyte/pull/14831) | Add future `start_date` and `end_date` validation                                                                                                                                                                                                                                                 |