🐛 Source Stripe: skip stream if more_perrmissions_rerquired

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

@@ -964,7 +964,7 @@
 - name: Stripe
   sourceDefinitionId: e094cb9a-26de-4645-8761-65c0c425d1de
   dockerRepository: airbyte/source-stripe
-  dockerImageTag: 0.1.36
+  dockerImageTag: 0.1.37
   documentationUrl: https://docs.airbyte.io/integrations/sources/stripe
   icon: stripe.svg
   sourceType: api

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

@@ -9498,7 +9498,7 @@
               type: "string"
               path_in_connector_config:
               - "client_secret"
-- dockerImage: "airbyte/source-stripe:0.1.36"
+- dockerImage: "airbyte/source-stripe:0.1.37"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/stripe"
     connectionSpecification:

airbyte-integrations/connectors/source-stripe/Dockerfile

@@ -12,5 +12,5 @@ COPY main.py ./
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.36
+LABEL io.airbyte.version=0.1.37
 LABEL io.airbyte.name=airbyte/source-stripe

airbyte-integrations/connectors/source-stripe/README.md

@@ -66,7 +66,8 @@ Customize `acceptance-test-config.yml` file to configure tests. See [Source Acce
 If your connector requires to create or destroy resources for use during acceptance tests create fixtures for it and place them inside integration_tests/acceptance.py.
 To run your integration tests with acceptance tests, from the connector root, run
 ```
-python -m pytest integration_tests -p integration_tests.acceptance
+docker build . --no-cache -t airbyte/source-stripe:dev \
+&& python -m pytest integration_tests -p integration_tests.acceptance
 ```
 
 ### Using gradle to run tests

airbyte-integrations/connectors/source-stripe/acceptance-test-config.yml

@@ -15,6 +15,7 @@ tests:
   basic_read:
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/full_refresh_configured_catalog.json"
+      empty_streams: ["external_account_bank_accounts"]
     # TEST 1 - Reading catalog without invoice_line_items
     - config_path: "secrets/config.json"
       configured_catalog_path: "integration_tests/non_invoice_line_items_catalog.json"

airbyte-integrations/connectors/source-stripe/integration_tests/configured_catalog.json

@@ -25,6 +25,17 @@
       "destination_sync_mode": "overwrite",
       "primary_key": [["id"]]
     },
+    {
+      "stream": {
+        "name": "external_account_bank_accounts",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite",
+      "primary_key": [["id"]]
+    },
     {
       "stream": {
         "name": "charges",

airbyte-integrations/connectors/source-stripe/integration_tests/full_refresh_configured_catalog.json

@@ -9,6 +9,17 @@
       "sync_mode": "full_refresh",
       "destination_sync_mode": "overwrite"
     },
+    {
+      "stream": {
+        "name": "external_account_bank_accounts",
+        "json_schema": {},
+        "supported_sync_modes": ["full_refresh"],
+        "source_defined_primary_key": [["id"]]
+      },
+      "sync_mode": "full_refresh",
+      "destination_sync_mode": "overwrite",
+      "primary_key": [["id"]]
+    },
     {
       "stream": {
         "name": "customer_balance_transactions",

airbyte-integrations/connectors/source-stripe/source_stripe/streams.py

@@ -12,6 +12,13 @@
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams.http import HttpStream
 
+STRIPE_ERROR_CODES: List = [
+    # stream requires additional permissions
+    "more_permissions_required",
+    # account_id doesn't have the access to the stream
+    "account_invalid",
+]
+
 
 class StripeStream(HttpStream, ABC):
     url_base = "https://api.stripe.com/v1/"
@@ -74,6 +81,30 @@ def stream_slices(
         for start, end in self.chunk_dates(self.start_date):
             yield {"created[gte]": start, "created[lte]": end}
 
+    def read_records(
+        self,
+        sync_mode: SyncMode,
+        cursor_field: List[str] = None,
+        stream_slice: Mapping[str, Any] = None,
+        stream_state: Mapping[str, Any] = None,
+    ) -> Iterable[Mapping[str, Any]]:
+        if stream_slice is None:
+            return []
+
+        try:
+            yield from super().read_records(sync_mode, cursor_field, stream_slice, stream_state)
+        except requests.exceptions.HTTPError as e:
+            status_code = e.response.status_code
+            parsed_error = e.response.json()
+            error_code = parsed_error.get("error", {}).get("code")
+            error_message = parsed_error.get("message")
+            # if the API Key doesn't have required permissions to particular stream, this stream will be skipped
+            if status_code == 403 and error_code in STRIPE_ERROR_CODES:
+                self.logger.warn(f"Stream {self.name} is skipped, due to {error_code}. Full message: {error_message}")
+                pass
+            else:
+                self.logger.error(f"Syncing stream {self.name} is failed, due to {error_code}. Full message: {error_message}")
+
 
 class SingleEmptySliceMixin(object):
     def stream_slices(
@@ -106,17 +137,6 @@ def get_updated_state(self, current_stream_state: MutableMapping[str, Any], late
         """
         return {self.cursor_field: max(latest_record.get(self.cursor_field), current_stream_state.get(self.cursor_field, 0))}
 
-    def read_records(
-        self,
-        sync_mode: SyncMode,
-        cursor_field: List[str] = None,
-        stream_slice: Mapping[str, Any] = None,
-        stream_state: Mapping[str, Any] = None,
-    ) -> Iterable[Mapping[str, Any]]:
-        if stream_slice is None:
-            return []
-        yield from super().read_records(sync_mode, cursor_field, stream_slice, stream_state)
-
     def stream_slices(
         self, *, sync_mode: SyncMode, cursor_field: List[str] = None, stream_state: Mapping[str, Any] = None
     ) -> Iterable[Optional[Mapping[str, Any]]]:

docs/integrations/sources/stripe.md

@@ -76,6 +76,7 @@ The Stripe connector should not run into Stripe API limitations under normal usa
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                                                |
 |:--------|:-----------|:---------------------------------------------------------|:-------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.1.37  | 2022-08-16 | [15686](https://github.com/airbytehq/airbyte/pull/15686) | Fixed the bug when the stream couldn't be fetched due to limited permission set, if so - it should be skipped                                                                                                                                 |
 | 0.1.36  | 2022-08-04 | [15292](https://github.com/airbytehq/airbyte/pull/15292) | Implement slicing                                                                                                                                      |
 | 0.1.35  | 2022-07-21 | [14924](https://github.com/airbytehq/airbyte/pull/14924) | Remove `additionalProperties` field from spec and schema                                                                                               |
 | 0.1.34  | 2022-07-01 | [14357](https://github.com/airbytehq/airbyte/pull/14357) | added external account streams -                                                                                                                       |