Merge remote-tracking branch 'origin/main' into baz/cdk/add-async-retriever-operation-auxiliary-requests

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -44,6 +44,7 @@
 from airbyte_cdk.sources.source import TState
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
+from airbyte_cdk.sources.streams.concurrent.abstract_stream_facade import AbstractStreamFacade
 from airbyte_cdk.sources.streams.concurrent.availability_strategy import (
     AlwaysAvailableAvailabilityStrategy,
 )
@@ -118,6 +119,12 @@ def __init__(
             message_repository=self.message_repository,
         )
 
+    # TODO: Remove this. This property is necessary to safely migrate Stripe during the transition state.
+    @property
+    def is_partially_declarative(self) -> bool:
+        """This flag used to avoid unexpected AbstractStreamFacade processing as concurrent streams."""
+        return False
+
     def read(
         self,
         logger: logging.Logger,
@@ -369,6 +376,14 @@ def _group_streams(
                     )
                 else:
                     synchronous_streams.append(declarative_stream)
+            # TODO: Remove this. This check is necessary to safely migrate Stripe during the transition state.
+            # Condition below needs to ensure that concurrent support is not lost for sources that already support
+            # it before migration, but now are only partially migrated to declarative implementation (e.g., Stripe).
+            elif (
+                isinstance(declarative_stream, AbstractStreamFacade)
+                and self.is_partially_declarative
+            ):
+                concurrent_streams.append(declarative_stream.get_underlying_stream())
             else:
                 synchronous_streams.append(declarative_stream)
 

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1490,7 +1490,11 @@ definitions:
       limit:
         title: Limit
         description: The maximum number of calls allowed within the interval.
-        type: integer
+        anyOf:
+          - type: integer
+          - type: string
+        interpolation_context:
+          - config
       interval:
         title: Interval
         description: The time interval for the rate limit.

airbyte_cdk/sources/declarative/decoders/composite_raw_decoder.py

@@ -107,6 +107,16 @@ class CsvParser(Parser):
     encoding: Optional[str] = "utf-8"
     delimiter: Optional[str] = ","
 
+    def _get_delimiter(self) -> Optional[str]:
+        """
+        Get delimiter from the configuration. Check for the escape character and decode it.
+        """
+        if self.delimiter is not None:
+            if self.delimiter.startswith("\\"):
+                self.delimiter = self.delimiter.encode("utf-8").decode("unicode_escape")
+
+        return self.delimiter
+
     def parse(
         self,
         data: BufferedIOBase,
@@ -115,8 +125,9 @@ def parse(
         Parse CSV data from decompressed bytes.
         """
         text_data = TextIOWrapper(data, encoding=self.encoding)  # type: ignore
-        reader = csv.DictReader(text_data, delimiter=self.delimiter or ",")
-        yield from reader
+        reader = csv.DictReader(text_data, delimiter=self._get_delimiter() or ",")
+        for row in reader:
+            yield row
 
 
 @dataclass

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -646,7 +646,7 @@ class Rate(BaseModel):
     class Config:
         extra = Extra.allow
 
-    limit: int = Field(
+    limit: Union[int, str] = Field(
         ...,
         description="The maximum number of calls allowed within the interval.",
         title="Limit",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -2091,10 +2091,10 @@ def create_dynamic_schema_loader(
     def create_json_decoder(model: JsonDecoderModel, config: Config, **kwargs: Any) -> Decoder:
         return JsonDecoder(parameters={})
 
-    @staticmethod
-    def create_csv_decoder(model: CsvDecoderModel, config: Config, **kwargs: Any) -> Decoder:
+    def create_csv_decoder(self, model: CsvDecoderModel, config: Config, **kwargs: Any) -> Decoder:
         return CompositeRawDecoder(
-            parser=ModelToComponentFactory._get_parser(model, config), stream_response=True
+            parser=ModelToComponentFactory._get_parser(model, config),
+            stream_response=False if self._emit_connector_builder_messages else True,
         )
 
     @staticmethod
@@ -2103,10 +2103,12 @@ def create_jsonl_decoder(model: JsonlDecoderModel, config: Config, **kwargs: Any
             parser=ModelToComponentFactory._get_parser(model, config), stream_response=True
         )
 
-    @staticmethod
-    def create_gzip_decoder(model: GzipDecoderModel, config: Config, **kwargs: Any) -> Decoder:
+    def create_gzip_decoder(
+        self, model: GzipDecoderModel, config: Config, **kwargs: Any
+    ) -> Decoder:
         return CompositeRawDecoder(
-            parser=ModelToComponentFactory._get_parser(model, config), stream_response=True
+            parser=ModelToComponentFactory._get_parser(model, config),
+            stream_response=False if self._emit_connector_builder_messages else True,
         )
 
     @staticmethod
@@ -3024,8 +3026,9 @@ def create_unlimited_call_rate_policy(
         )
 
     def create_rate(self, model: RateModel, config: Config, **kwargs: Any) -> Rate:
+        interpolated_limit = InterpolatedString.create(str(model.limit), parameters={})
         return Rate(
-            limit=model.limit,
+            limit=int(interpolated_limit.eval(config=config)),
             interval=parse_duration(model.interval),
         )
 

unit_tests/sources/declarative/decoders/test_composite_decoder.py

@@ -62,7 +62,9 @@ def test_composite_raw_decoder_gzip_csv_parser(requests_mock, encoding: str):
     )
     response = requests.get("https://airbyte.io/", stream=True)
 
-    parser = GzipParser(inner_parser=CsvParser(encoding=encoding, delimiter="\t"))
+    # the delimiter is set to `\\t` intentionally to test the parsing logic here
+    parser = GzipParser(inner_parser=CsvParser(encoding=encoding, delimiter="\\t"))
+
     composite_raw_decoder = CompositeRawDecoder(parser=parser)
     counter = 0
     for _ in composite_raw_decoder.decode(response):

unit_tests/sources/declarative/requesters/test_http_requester.py

@@ -946,7 +946,7 @@ def test_backoff_strategy_from_manifest_is_respected(http_requester_factory: Any
     )
 
 
-def test_http_requester_with_mock_apibudget(http_requester_factory, monkeypatch):
+def test_http_requester_with_mock_api_budget(http_requester_factory, monkeypatch):
     mock_budget = MagicMock(spec=HttpAPIBudget)
 
     requester = http_requester_factory(