feat(source-stripe): add convert to concurrent stream

airbyte-integrations/connectors/source-stripe/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: e094cb9a-26de-4645-8761-65c0c425d1de
-  dockerImageTag: 5.6.0
+  dockerImageTag: 5.7.0
   dockerRepository: airbyte/source-stripe
   documentationUrl: https://docs.airbyte.com/integrations/sources/stripe
   erdUrl: https://dbdocs.io/airbyteio/source-stripe?view=relationships

airbyte-integrations/connectors/source-stripe/pyproject.toml

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "5.6.0"
+version = "5.7.0"
 name = "source-stripe"
 description = "Source implementation for Stripe."
 authors = [ "Airbyte <contact@airbyte.io>",]

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

@@ -538,54 +538,30 @@ def streams(self, config: MutableMapping[str, Any]) -> List[Stream]:
         ]
 
         state_manager = ConnectorStateManager(state=self._state)
+        converter = EpochValueConcurrentStreamStateConverter()
+        start = datetime.fromtimestamp(self._start_date_to_timestamp(config), timezone.utc)
+        slice_range = timedelta(days=config["slice_range"])
+
         return [
-            self._to_concurrent(
+            self.convert_to_concurrent_stream(
+                logger,
                 stream,
-                datetime.fromtimestamp(self._start_date_to_timestamp(config), timezone.utc),
-                timedelta(days=config["slice_range"]),
                 state_manager,
+                cursor=self.initialize_cursor(
+                    stream,
+                    state_manager,
+                    converter,
+                    self._SLICE_BOUNDARY_FIELDS_BY_IMPLEMENTATION.get(type(stream)),
+                    start,
+                    converter.get_end_provider(),
+                    slice_range=slice_range,
+                ),
             )
+            if self._SLICE_BOUNDARY_FIELDS_BY_IMPLEMENTATION.get(type(stream))
+            else stream
             for stream in streams
         ]
 
-    def _to_concurrent(
-        self, stream: Stream, fallback_start: datetime, slice_range: timedelta, state_manager: ConnectorStateManager
-    ) -> Stream:
-        if stream.name in self._streams_configured_as_full_refresh:
-            return StreamFacade.create_from_stream(
-                stream,
-                self,
-                entrypoint_logger,
-                self._create_empty_state(),
-                FinalStateCursor(stream_name=stream.name, stream_namespace=stream.namespace, message_repository=self.message_repository),
-            )
-
-        state = state_manager.get_stream_state(stream.name, stream.namespace)
-        slice_boundary_fields = self._SLICE_BOUNDARY_FIELDS_BY_IMPLEMENTATION.get(type(stream))
-        if slice_boundary_fields:
-            cursor_field = CursorField(stream.cursor_field) if isinstance(stream.cursor_field, str) else CursorField(stream.cursor_field[0])
-            converter = EpochValueConcurrentStreamStateConverter()
-            cursor = ConcurrentCursor(
-                stream.name,
-                stream.namespace,
-                state_manager.get_stream_state(stream.name, stream.namespace),
-                self.message_repository,
-                state_manager,
-                converter,
-                cursor_field,
-                slice_boundary_fields,
-                fallback_start,
-                converter.get_end_provider(),
-                timedelta(seconds=0),
-                slice_range,
-            )
-            return StreamFacade.create_from_stream(stream, self, entrypoint_logger, state, cursor)
-
-        return stream
-
-    def _create_empty_state(self) -> MutableMapping[str, Any]:
-        return {}
-
     @staticmethod
     def _start_date_to_timestamp(config: Mapping[str, Any]) -> int:
         if "start_date" not in config:

airbyte-integrations/connectors/source-stripe/unit_tests/integration/test_persons.py

@@ -442,7 +442,8 @@ def test_rate_limit_max_attempts_exceeded(self, http_mocker: HttpMocker) -> None
                 FailureType.system_error,
                 FailureType.config_error,
             ]
-            assert "Request rate limit exceeded" in actual_messages.errors[0].trace.error.internal_message
+
+            assert "Request rate limit exceeded" in actual_messages.errors[0].trace.error.message
 
     @HttpMocker()
     def test_incremental_rate_limit_max_attempts_exceeded(self, http_mocker: HttpMocker) -> None:

airbyte-integrations/connectors/source-stripe/unit_tests/test_source.py

@@ -74,9 +74,7 @@ def test_when_streams_return_full_refresh_as_concurrent():
         _NO_STATE,
     ).streams(_a_valid_config())
 
-    # bank_accounts (as it is defined as full_refresh)
-    # balance_transactions, events, files, file_links and shipping_rates (as it is always concurrent now)
-    assert len(list(filter(lambda stream: isinstance(stream, StreamFacade), streams))) == 6
+    assert len(list(filter(lambda stream: isinstance(stream, StreamFacade), streams))) == 5
 
 
 @pytest.mark.parametrize(

docs/integrations/sources/stripe.md

@@ -239,6 +239,7 @@ Each record is marked with `is_deleted` flag when the appropriate event happens
 
 | Version | Date       | Pull Request                                              | Subject                                                                                                                                                                                                                       |
 |:--------|:-----------|:----------------------------------------------------------|:------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 5.7.0   | 2024-10-10 | [45701](https://github.com/airbytehq/airbyte/pull/45701)  | Update source to use `convert_to_concurrent_stream`                                                                                                                                                                           |
 | 5.6.0   | 2024-09-10 | [44891](https://github.com/airbytehq/airbyte/pull/44891)  | Update `Payment Methods` stream                                                                                                                                                                                               |
 | 5.5.4   | 2024-09-09 | [45348](https://github.com/airbytehq/airbyte/pull/45348)  | Remove `stripe` python package                                                                                                                                                                                                |
 | 5.5.3   | 2024-09-03 | [45101](https://github.com/airbytehq/airbyte/pull/45101)  | Fix regression following pagination issue fix                                                                                                                                                                                 |