Merge branch 'main' into aj/feat/mini-cat-test-suites

Author: aaronsteers

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2307,6 +2307,27 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
+  KeyTransformation:
+    title: Transformation to apply for extracted object keys by Dpath Flatten Fields
+    type: object
+    required:
+      - type
+    properties:
+      type:
+        type: string
+        enum: [ KeyTransformation ]
+      prefix:
+        title: Key Prefix
+        description: Prefix to add for object keys. If not provided original keys remain unchanged.
+        type: string
+        examples:
+          - flattened_
+      suffix:
+        title: Key Suffix
+        description: Suffix to add for object keys. If not provided original keys remain unchanged.
+        type: string
+        examples:
+          - _flattened
   DpathFlattenFields:
     title: Dpath Flatten Fields
     description: A transformation that flatten field values to the to top of the record.
@@ -2335,6 +2356,11 @@ definitions:
         title: Replace Origin Record
         description: Whether to replace the origin record or not. Default is False.
         type: boolean
+      key_transformation:
+        title: Key transformation
+        description: Transformation for object keys. If not provided, original key will be used.
+        type: object
+        "$ref": "#/definitions/KeyTransformation"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -879,6 +879,25 @@ class FlattenFields(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
+class KeyTransformation(BaseModel):
+    prefix: Optional[Union[str, None]] = Field(
+        None,
+        description="Prefix to add for object keys. If not provided original keys remain unchanged.",
+        examples=[
+            "flattened_",
+        ],
+        title="Key Prefix",
+    )
+    suffix: Optional[Union[str, None]] = Field(
+        None,
+        description="Suffix to add for object keys. If not provided original keys remain unchanged.",
+        examples=[
+            "_flattened",
+        ],
+        title="Key Suffix",
+    )
+
+
 class DpathFlattenFields(BaseModel):
     type: Literal["DpathFlattenFields"]
     field_path: List[str] = Field(
@@ -897,6 +916,11 @@ class DpathFlattenFields(BaseModel):
         description="Whether to replace the origin record or not. Default is False.",
         title="Replace Origin Record",
     )
+    key_transformation: Optional[Union[KeyTransformation, None]] = Field(
+        None,
+        description="Transformation for object keys. If not provided, original key will be used.",
+        title="Key transformation",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -498,6 +498,7 @@
 from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition
 from airbyte_cdk.sources.declarative.transformations.dpath_flatten_fields import (
     DpathFlattenFields,
+    KeyTransformation,
 )
 from airbyte_cdk.sources.declarative.transformations.flatten_fields import (
     FlattenFields,
@@ -790,13 +791,24 @@ def create_dpath_flatten_fields(
         self, model: DpathFlattenFieldsModel, config: Config, **kwargs: Any
     ) -> DpathFlattenFields:
         model_field_path: List[Union[InterpolatedString, str]] = [x for x in model.field_path]
+        key_transformation = (
+            KeyTransformation(
+                config=config,
+                prefix=model.key_transformation.prefix,
+                suffix=model.key_transformation.suffix,
+                parameters=model.parameters or {},
+            )
+            if model.key_transformation is not None
+            else None
+        )
         return DpathFlattenFields(
             config=config,
             field_path=model_field_path,
             delete_origin_value=model.delete_origin_value
             if model.delete_origin_value is not None
             else False,
             replace_record=model.replace_record if model.replace_record is not None else False,
+            key_transformation=key_transformation,
             parameters=model.parameters or {},
         )
 
@@ -2054,6 +2066,7 @@ def create_default_paginator(
         config: Config,
         *,
         url_base: str,
+        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
         decoder: Optional[Decoder] = None,
         cursor_used_for_stop_condition: Optional[DeclarativeCursor] = None,
     ) -> Union[DefaultPaginator, PaginatorTestReadDecorator]:
@@ -2075,7 +2088,10 @@ def create_default_paginator(
             else None
         )
         pagination_strategy = self._create_component_from_model(
-            model=model.pagination_strategy, config=config, decoder=decoder_to_use
+            model=model.pagination_strategy,
+            config=config,
+            decoder=decoder_to_use,
+            extractor_model=extractor_model,
         )
         if cursor_used_for_stop_condition:
             pagination_strategy = StopConditionPaginationStrategyDecorator(
@@ -2572,7 +2588,12 @@ def create_oauth_authenticator(
         )
 
     def create_offset_increment(
-        self, model: OffsetIncrementModel, config: Config, decoder: Decoder, **kwargs: Any
+        self,
+        model: OffsetIncrementModel,
+        config: Config,
+        decoder: Decoder,
+        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
+        **kwargs: Any,
     ) -> OffsetIncrement:
         if isinstance(decoder, PaginationDecoderDecorator):
             inner_decoder = decoder.decoder
@@ -2587,10 +2608,24 @@ def create_offset_increment(
                 self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(inner_decoder))
             )
 
+        # Ideally we would instantiate the runtime extractor from highest most level (in this case the SimpleRetriever)
+        # so that it can be shared by OffSetIncrement and RecordSelector. However, due to how we instantiate the
+        # decoder with various decorators here, but not in create_record_selector, it is simpler to retain existing
+        # behavior by having two separate extractors with identical behavior since they use the same extractor model.
+        # When we have more time to investigate we can look into reusing the same component.
+        extractor = (
+            self._create_component_from_model(
+                model=extractor_model, config=config, decoder=decoder_to_use
+            )
+            if extractor_model
+            else None
+        )
+
         return OffsetIncrement(
             page_size=model.page_size,
             config=config,
             decoder=decoder_to_use,
+            extractor=extractor,
             inject_on_first_request=model.inject_on_first_request or False,
             parameters=model.parameters or {},
         )
@@ -2954,6 +2989,7 @@ def create_simple_retriever(
                 model=model.paginator,
                 config=config,
                 url_base=url_base,
+                extractor_model=model.record_selector.extractor,
                 decoder=decoder,
                 cursor_used_for_stop_condition=cursor_used_for_stop_condition,
             )

airbyte_cdk/sources/declarative/requesters/paginators/strategies/offset_increment.py

@@ -12,6 +12,7 @@
     JsonDecoder,
     PaginationDecoderDecorator,
 )
+from airbyte_cdk.sources.declarative.extractors.dpath_extractor import RecordExtractor
 from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 from airbyte_cdk.sources.declarative.requesters.paginators.strategies.pagination_strategy import (
     PaginationStrategy,
@@ -46,6 +47,7 @@ class OffsetIncrement(PaginationStrategy):
     config: Config
     page_size: Optional[Union[str, int]]
     parameters: InitVar[Mapping[str, Any]]
+    extractor: Optional[RecordExtractor]
     decoder: Decoder = field(
         default_factory=lambda: PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
     )
@@ -75,6 +77,14 @@ def next_page_token(
     ) -> Optional[Any]:
         decoded_response = next(self.decoder.decode(response))
 
+        if self.extractor:
+            page_size_from_response = len(list(self.extractor.extract_records(response=response)))
+            # The extractor could return 0 records which is valid, but evaluates to False. Our fallback in other
+            # cases as the best effort option is to use the incoming last_page_size
+            last_page_size = (
+                page_size_from_response if page_size_from_response is not None else last_page_size
+            )
+
         # Stop paginating when there are fewer records than the page size or the current page has no records
         if (
             self._page_size

airbyte_cdk/sources/declarative/transformations/dpath_flatten_fields.py

@@ -8,6 +8,24 @@
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
 
+@dataclass
+class KeyTransformation:
+    config: Config
+    parameters: InitVar[Mapping[str, Any]]
+    prefix: Optional[str] = None
+    suffix: Optional[str] = None
+
+    def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        if self.prefix is not None:
+            self.prefix = InterpolatedString.create(self.prefix, parameters=parameters).eval(
+                self.config
+            )
+        if self.suffix is not None:
+            self.suffix = InterpolatedString.create(self.suffix, parameters=parameters).eval(
+                self.config
+            )
+
+
 @dataclass
 class DpathFlattenFields(RecordTransformation):
     """
@@ -16,6 +34,7 @@ class DpathFlattenFields(RecordTransformation):
     field_path: List[Union[InterpolatedString, str]] path to the field to flatten.
     delete_origin_value: bool = False whether to delete origin field or keep it. Default is False.
     replace_record: bool = False whether to replace origin record or not. Default is False.
+    key_transformation: KeyTransformation = None how to transform extracted object keys
 
     """
 
@@ -24,17 +43,35 @@ class DpathFlattenFields(RecordTransformation):
     parameters: InitVar[Mapping[str, Any]]
     delete_origin_value: bool = False
     replace_record: bool = False
+    key_transformation: Optional[KeyTransformation] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
+        self._parameters = parameters
         self._field_path = [
-            InterpolatedString.create(path, parameters=parameters) for path in self.field_path
+            InterpolatedString.create(path, parameters=self._parameters) for path in self.field_path
         ]
         for path_index in range(len(self.field_path)):
             if isinstance(self.field_path[path_index], str):
                 self._field_path[path_index] = InterpolatedString.create(
-                    self.field_path[path_index], parameters=parameters
+                    self.field_path[path_index], parameters=self._parameters
                 )
 
+    def _apply_key_transformation(self, extracted: Mapping[str, Any]) -> Mapping[str, Any]:
+        if self.key_transformation:
+            if self.key_transformation.prefix:
+                extracted = {
+                    f"{self.key_transformation.prefix}{key}": value
+                    for key, value in extracted.items()
+                }
+
+            if self.key_transformation.suffix:
+                extracted = {
+                    f"{key}{self.key_transformation.suffix}": value
+                    for key, value in extracted.items()
+                }
+
+        return extracted
+
     def transform(
         self,
         record: Dict[str, Any],
@@ -50,6 +87,8 @@ def transform(
             extracted = dpath.get(record, path, default=[])
 
         if isinstance(extracted, dict):
+            extracted = self._apply_key_transformation(extracted)
+
             if self.replace_record and extracted:
                 dpath.delete(record, "**")
                 record.update(extracted)

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -66,6 +66,7 @@
 from airbyte_cdk.sources.declarative.models import DatetimeBasedCursor as DatetimeBasedCursorModel
 from airbyte_cdk.sources.declarative.models import DeclarativeStream as DeclarativeStreamModel
 from airbyte_cdk.sources.declarative.models import DefaultPaginator as DefaultPaginatorModel
+from airbyte_cdk.sources.declarative.models import DpathExtractor as DpathExtractorModel
 from airbyte_cdk.sources.declarative.models import (
     GroupingPartitionRouter as GroupingPartitionRouterModel,
 )
@@ -1831,6 +1832,7 @@ def test_create_default_paginator():
         component_definition=paginator_manifest,
         config=input_config,
         url_base="https://airbyte.io",
+        extractor_model=DpathExtractor(field_path=["results"], config=input_config, parameters={}),
         decoder=JsonDecoder(parameters={}),
     )
 
@@ -1968,6 +1970,7 @@ def test_create_default_paginator():
             DefaultPaginator(
                 pagination_strategy=OffsetIncrement(
                     page_size=10,
+                    extractor=None,
                     config={"apikey": "verysecrettoken", "repos": ["airbyte", "airbyte-cloud"]},
                     parameters={},
                 ),
@@ -2641,18 +2644,31 @@ def test_create_offset_increment():
         page_size=10,
         inject_on_first_request=True,
     )
+
+    expected_extractor = DpathExtractor(field_path=["results"], config=input_config, parameters={})
+    extractor_model = DpathExtractorModel(
+        type="DpathExtractor", field_path=expected_extractor.field_path
+    )
+
     expected_strategy = OffsetIncrement(
-        page_size=10, inject_on_first_request=True, parameters={}, config=input_config
+        page_size=10,
+        inject_on_first_request=True,
+        extractor=expected_extractor,
+        parameters={},
+        config=input_config,
     )
 
     strategy = factory.create_offset_increment(
-        model, input_config, decoder=JsonDecoder(parameters={})
+        model, input_config, extractor_model=extractor_model, decoder=JsonDecoder(parameters={})
     )
 
     assert strategy.page_size == expected_strategy.page_size
     assert strategy.inject_on_first_request == expected_strategy.inject_on_first_request
     assert strategy.config == input_config
 
+    assert isinstance(strategy.extractor, DpathExtractor)
+    assert strategy.extractor.field_path == expected_extractor.field_path
+
 
 class MyCustomSchemaLoader(SchemaLoader):
     def get_json_schema(self) -> Mapping[str, Any]: