source-facebook-marketing: allow configuration of MAX_BATCH_SIZE

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

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

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

@@ -1827,7 +1827,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-facebook-marketing:0.2.53"
+- dockerImage: "airbyte/source-facebook-marketing:0.2.54"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/facebook-marketing"
     changelogUrl: "https://docs.airbyte.io/integrations/sources/facebook-marketing"
@@ -2159,6 +2159,15 @@
           mininum: 1
           exclusiveMinimum: 0
           type: "integer"
+        max_batch_size:
+          title: "Maximum size of Batched Requests"
+          description: "Maximum batch size used when sending batch requests to Facebook\
+            \ API. Most users do not need to set this field unless they specifically\
+            \ need to tune   the connector to address specific issues or use cases."
+          default: 50
+          order: 9
+          exclusiveMinimum: 0
+          type: "integer"
       required:
       - "account_id"
       - "start_date"

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

airbyte-integrations/connectors/source-facebook-marketing/integration_tests/spec.json

@@ -320,6 +320,14 @@
         "mininum": 1,
         "exclusiveMinimum": 0,
         "type": "integer"
+      },
+      "max_batch_size": {
+        "title": "Maximum size of Batched Requests",
+        "description": "Maximum batch size used when sending batch requests to Facebook API. Most users do not need to set this field unless they specifically need to tune the connector to address specific issues or use cases.",
+        "default": 50,
+        "order": 9,
+        "exclusiveMinimum": 0,
+        "type": "integer"
       }
     },
     "required": ["account_id", "start_date", "access_token"]

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

@@ -71,49 +71,60 @@ def streams(self, config: Mapping[str, Any]) -> List[Type[Stream]]:
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
             ),
             Ads(
                 api=api,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
             ),
-            AdCreatives(api=api, fetch_thumbnail_images=config.fetch_thumbnail_images, page_size=config.page_size),
-            AdsInsights(page_size=config.page_size, **insights_args),
-            AdsInsightsAgeAndGender(page_size=config.page_size, **insights_args),
-            AdsInsightsCountry(page_size=config.page_size, **insights_args),
-            AdsInsightsRegion(page_size=config.page_size, **insights_args),
-            AdsInsightsDma(page_size=config.page_size, **insights_args),
-            AdsInsightsPlatformAndDevice(page_size=config.page_size, **insights_args),
-            AdsInsightsActionType(page_size=config.page_size, **insights_args),
+            AdCreatives(
+                api=api,
+                fetch_thumbnail_images=config.fetch_thumbnail_images,
+                page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
+            ),
+            AdsInsights(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
+            AdsInsightsAgeAndGender(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
+            AdsInsightsCountry(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
+            AdsInsightsRegion(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
+            AdsInsightsDma(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
+            AdsInsightsPlatformAndDevice(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
+            AdsInsightsActionType(page_size=config.page_size, max_batch_size=config.max_batch_size, **insights_args),
             Campaigns(
                 api=api,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
             ),
             Images(
                 api=api,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
             ),
             Videos(
                 api=api,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
             ),
             Activities(
                 api=api,
                 start_date=config.start_date,
                 end_date=config.end_date,
                 include_deleted=config.include_deleted,
                 page_size=config.page_size,
+                max_batch_size=config.max_batch_size,
             ),
         ]
 

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

@@ -172,3 +172,10 @@ class Config:
         mininum=1,
         default=28,
     )
+
+    max_batch_size: Optional[PositiveInt] = Field(
+        title="Maximum size of Batched Requests",
+        order=9,
+        description="Maximum batch size used when sending batch requests to Facebook API. Most users do not need to set this field unless they specifically need to tune the connector to address specific issues or use cases.",
+        default=50,
+    )

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

@@ -16,7 +16,7 @@
 from facebook_business.adobjects.adimage import AdImage
 from facebook_business.api import FacebookAdsApiBatch, FacebookRequest, FacebookResponse
 
-from .common import MAX_BATCH_SIZE, deep_merge
+from .common import deep_merge
 
 if TYPE_CHECKING:  # pragma: no cover
     from source_facebook_marketing.api import API
@@ -37,11 +37,12 @@ class FBMarketingStream(Stream, ABC):
     # entity prefix for `include_deleted` filter, it usually matches singular version of stream name
     entity_prefix = None
 
-    def __init__(self, api: "API", include_deleted: bool = False, page_size: int = 100, **kwargs):
+    def __init__(self, api: "API", include_deleted: bool = False, page_size: int = 100, max_batch_size: int = 50, **kwargs):
         super().__init__(**kwargs)
         self._api = api
         self.page_size = page_size if page_size is not None else 100
         self._include_deleted = include_deleted if self.enable_deleted else False
+        self.max_batch_size = max_batch_size if max_batch_size is not None else 50
 
     @cached_property
     def fields(self) -> List[str]:
@@ -68,7 +69,7 @@ def failure(response: FacebookResponse):
         api_batch: FacebookAdsApiBatch = self._api.api.new_batch()
         for request in pending_requests:
             api_batch.add_request(request, success=success, failure=failure)
-            if len(api_batch) == MAX_BATCH_SIZE:
+            if len(api_batch) == self.max_batch_size:
                 self._execute_batch(api_batch)
                 yield from records
                 records = []

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

@@ -19,7 +19,6 @@
 FACEBOOK_UNKNOWN_ERROR_CODE = 99
 FACEBOOK_CONNECTION_RESET_ERROR_CODE = 104
 DEFAULT_SLEEP_INTERVAL = pendulum.duration(minutes=1)
-MAX_BATCH_SIZE = 50
 
 logger = logging.getLogger("airbyte")
 

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

@@ -11,7 +11,6 @@
 from facebook_business.api import FacebookAdsApi, FacebookAdsApiBatch, FacebookRequest
 from source_facebook_marketing.api import MyFacebookAdsApi
 from source_facebook_marketing.streams.base_streams import FBMarketingStream
-from source_facebook_marketing.streams.common import MAX_BATCH_SIZE
 
 
 @pytest.fixture(name="mock_batch_responses")
@@ -64,7 +63,7 @@ def test_execute_in_batch_with_many_requests(self, api, batch, mock_batch_respon
         )
 
         stream = SomeTestStream(api=api)
-        requests = [FacebookRequest("node", "GET", "endpoint") for _ in range(MAX_BATCH_SIZE + 1)]
+        requests = [FacebookRequest("node", "GET", "endpoint") for _ in range(50 + 1)]
 
         result = list(stream.execute_in_batch(requests))
 

docs/integrations/sources/facebook-marketing.md

@@ -120,6 +120,7 @@ Please be informed that the connector uses the `lookback_window` parameter to pe
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                                                                                                                                                                                           |
 |:--------|:-----------|:---------------------------------------------------------|:--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.2.54  | 2022-06-29 | [14267](https://github.com/airbytehq/airbyte/pull/14267) | Make MAX_BATCH_SIZE available in config |                     
 | 0.2.53  | 2022-06-16 | [13623](https://github.com/airbytehq/airbyte/pull/13623) | Add fields `bid_amount` `bid_strategy` `bid_constraints` to `ads_set` stream                                                                                                                                                                                                                      |
 | 0.2.52  | 2022-06-14 | [13749](https://github.com/airbytehq/airbyte/pull/13749) | Fix the `not syncing any data` issue                                                                                                                                                                                                                                                              |
 | 0.2.51  | 2022-05-30 | [13317](https://github.com/airbytehq/airbyte/pull/13317) | Change tax_id to string (Canadian has letter in tax_id)                                                                                                                                                                                                                                           |