Update to complex type resolving

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1800,30 +1800,19 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
-  SchemaFieldType:
+  ComplexFieldType:
     title: Schema Field Type
-    description: (This component is experimental. Use at your own risk.) Represents a mapping between a current type and its corresponding target type for property.
+    description: (This component is experimental. Use at your own risk.) Represents a complex field type.
     type: object
     required:
       - field_type
     properties:
       field_type:
         type: string
       items:
-        "$ref": "#/definitions/SchemaFieldType"
-  ItemsTypeMap:
-    title: Types Map
-    description: (This component is experimental. Use at your own risk.) Represents a mapping between a current type and its corresponding target type for property.
-    type: object
-    required:
-      - target_type
-    properties:
-      condition:
-        type: string
-        interpolation_context:
-          - raw_schema
-      target_type:
-          "$ref": "#/definitions/SchemaFieldType"
+        anyOf:
+          - type: string
+          - "$ref": "#/definitions/ComplexFieldType"
   TypesMap:
     title: Types Map
     description: (This component is experimental. Use at your own risk.) Represents a mapping between a current type and its corresponding target type.
@@ -1838,6 +1827,7 @@ definitions:
           - type: array
             items:
               type: string
+          - "$ref": "#/definitions/ComplexFieldType"
       current_type:
         anyOf:
           - type: string
@@ -1848,8 +1838,6 @@ definitions:
         type: string
         interpolation_context:
           - raw_schema
-      items_type:
-        "$ref": "#/definitions/ItemsTypeMap"
   SchemaTypeIdentifier:
     title: Schema Type Identifier
     description: (This component is experimental. Use at your own risk.) Identifies schema details for dynamic schema extraction and processing.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -738,21 +738,15 @@ class HttpResponseFilter(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
-class SchemaFieldType(BaseModel):
+class ComplexFieldType(BaseModel):
     field_type: str
-    items: Optional[SchemaFieldType] = None
-
-
-class ItemsTypeMap(BaseModel):
-    condition: Optional[str] = None
-    target_type: SchemaFieldType
+    items: Optional[Union[str, ComplexFieldType]] = None
 
 
 class TypesMap(BaseModel):
-    target_type: Union[str, List[str]]
+    target_type: Union[str, List[str], ComplexFieldType]
     current_type: Union[str, List[str]]
     condition: Optional[str] = None
-    items_type: Optional[ItemsTypeMap] = None
 
 
 class SchemaTypeIdentifier(BaseModel):
@@ -2297,7 +2291,7 @@ class DynamicDeclarativeStream(BaseModel):
     )
 
 
-SchemaFieldType.update_forward_refs()
+ComplexFieldType.update_forward_refs()
 CompositeErrorHandler.update_forward_refs()
 DeclarativeSource1.update_forward_refs()
 DeclarativeSource2.update_forward_refs()

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -245,7 +245,7 @@
     InlineSchemaLoader as InlineSchemaLoaderModel,
 )
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    ItemsTypeMap as ItemsTypeMapModel,
+    ComplexFieldType as ComplexFieldTypeModel,
 )
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     IterableDecoder as IterableDecoderModel,
@@ -435,7 +435,7 @@
     DefaultSchemaLoader,
     DynamicSchemaLoader,
     InlineSchemaLoader,
-    ItemsTypeMap,
+    ComplexFieldType,
     JsonFileSchemaLoader,
     SchemaTypeIdentifier,
     TypesMap,
@@ -576,7 +576,7 @@ def _init_mappings(self) -> None:
             DynamicSchemaLoaderModel: self.create_dynamic_schema_loader,
             SchemaTypeIdentifierModel: self.create_schema_type_identifier,
             TypesMapModel: self.create_types_map,
-            ItemsTypeMapModel: self.create_items_type_map,
+            ComplexFieldTypeModel: self.create_complex_field_type,
             JwtAuthenticatorModel: self.create_jwt_authenticator,
             LegacyToPerPartitionStateMigrationModel: self.create_legacy_to_per_partition_state_migration,
             ListPartitionRouterModel: self.create_list_partition_router,
@@ -1899,27 +1899,28 @@ def create_inline_schema_loader(
     ) -> InlineSchemaLoader:
         return InlineSchemaLoader(schema=model.schema_ or {}, parameters={})
 
-    def create_items_type_map(
-        self, model: ItemsTypeMapModel, config: Config, **kwargs: Any
-    ) -> ItemsTypeMap:
-        type_mapping = self._create_component_from_model(model=model.type_mapping, config=config)
-        model_items_type_pointer: List[Union[InterpolatedString, str]] = (
-            [x for x in model.items_type_pointer] if model.items_type_pointer else []
+    def create_complex_field_type(
+        self, model: ComplexFieldTypeModel, config: Config, **kwargs: Any
+    ) -> ComplexFieldType:
+        items = (
+            self._create_component_from_model(model=model.items, config=config)
+            if isinstance(model.items, ComplexFieldTypeModel)
+            else model.items
         )
-        return ItemsTypeMap(items_type_pointer=model_items_type_pointer, type_mapping=type_mapping)
+
+        return ComplexFieldType(field_type=model.field_type, items=items)
 
     def create_types_map(self, model: TypesMapModel, config: Config, **kwargs: Any) -> TypesMap:
-        items_type = (
-            self._create_component_from_model(model=model.items_type, config=config)
-            if isinstance(model.items_type, ItemsTypeMapModel)
-            else model.items_type
+        target_type = (
+            self._create_component_from_model(model=model.target_type, config=config)
+            if isinstance(model.target_type, ComplexFieldTypeModel)
+            else model.target_type
         )
 
         return TypesMap(
-            target_type=model.target_type,
+            target_type=target_type,
             current_type=model.current_type,
             condition=model.condition if model.condition is not None else "True",
-            items_type=items_type,
         )
 
     def create_schema_type_identifier(

airbyte_cdk/sources/declarative/schema/__init__.py

@@ -5,8 +5,8 @@
 from airbyte_cdk.sources.declarative.schema.default_schema_loader import DefaultSchemaLoader
 from airbyte_cdk.sources.declarative.schema.dynamic_schema_loader import (
     DynamicSchemaLoader,
-    ItemsTypeMap,
     SchemaTypeIdentifier,
+    ComplexFieldType,
     TypesMap,
 )
 from airbyte_cdk.sources.declarative.schema.inline_schema_loader import InlineSchemaLoader
@@ -19,7 +19,7 @@
     "SchemaLoader",
     "InlineSchemaLoader",
     "DynamicSchemaLoader",
+    "ComplexFieldType",
     "TypesMap",
-    "ItemsTypeMap",
     "SchemaTypeIdentifier",
 ]

airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py

@@ -18,7 +18,7 @@
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.types import Config, StreamSlice, StreamState
 
-AIRBYTE_DATA_TYPES: Mapping[str, Mapping[str, Any]] = {
+AIRBYTE_DATA_TYPES: Mapping[str, MutableMapping[str, Any]] = {
     "string": {"type": ["null", "string"]},
     "boolean": {"type": ["null", "boolean"]},
     "date": {"type": ["null", "string"], "format": "date"},
@@ -45,14 +45,33 @@
 }
 
 
+@deprecated("This class is experimental. Use at your own risk.", category=ExperimentalClassWarning)
+@dataclass(frozen=True)
+class ComplexFieldType:
+    """
+    Identifies complex field type
+    """
+
+    field_type: str
+    items: Optional[Union[str, "ComplexFieldType"]] = None
+
+    def __post_init__(self) -> None:
+        """
+        Enforces that `items` is only used when `field_type` is a array
+        """
+        # `items_type` is valid only for array target types
+        if self.items and self.field_type != "array":
+            raise ValueError("'items' can only be used when 'field_type' is an array.")
+
+
 @deprecated("This class is experimental. Use at your own risk.", category=ExperimentalClassWarning)
 @dataclass(frozen=True)
 class TypesMap:
     """
     Represents a mapping between a current type and its corresponding target type.
     """
 
-    target_type: Union[List[str], str]
+    target_type: Union[List[str], str, ComplexFieldType]
     current_type: Union[List[str], str]
     condition: Optional[str]
 
@@ -188,13 +207,39 @@ def _get_type(
             first_type = self._get_airbyte_type(mapped_field_type[0])
             second_type = self._get_airbyte_type(mapped_field_type[1])
             return {"oneOf": [first_type, second_type]}
+
         elif isinstance(mapped_field_type, str):
             return self._get_airbyte_type(mapped_field_type)
+
+        elif isinstance(mapped_field_type, ComplexFieldType):
+            return self._resolve_complex_type(mapped_field_type)
+
         else:
             raise ValueError(
                 f"Invalid data type. Available string or two items list of string. Got {mapped_field_type}."
             )
 
+    def _resolve_complex_type(self, complex_type: ComplexFieldType) -> Mapping[str, Any]:
+        types = [complex_type]
+        resolved_type = {}
+
+        while types:
+            current_type = types.pop()
+            if not current_type.items:
+                resolved_type = self._get_airbyte_type(current_type.field_type)
+            else:
+                field_type = self._get_airbyte_type(current_type.field_type)
+
+                if isinstance(current_type.items, str):
+                    items_type = current_type.items
+                else:
+                    types.append(current_type.items)
+                    continue  # Skip the following lines until the stack is resolved
+                field_type["items"] = self._get_airbyte_type(items_type)
+                resolved_type = field_type
+
+        return resolved_type
+
     def _replace_type_if_not_valid(
         self,
         field_type: Union[List[str], str],
@@ -216,7 +261,7 @@ def _replace_type_if_not_valid(
         return field_type
 
     @staticmethod
-    def _get_airbyte_type(field_type: str) -> Mapping[str, Any]:
+    def _get_airbyte_type(field_type: str) -> MutableMapping[str, Any]:
         """
         Maps a field type to its corresponding Airbyte type definition.
         """

unit_tests/sources/declarative/schema/test_dynamic_schema_loader.py

@@ -85,17 +85,16 @@
                     "key_pointer": ["name"],
                     "type_pointer": ["type"],
                     "types_mapping": [
-                        {"target_type": "string", "current_type": "singleLineText"},
                         {
-                            "target_type": "array",
-                            "current_type": "formula",
-                            "items_type": {
-                                "items_type_pointer": ["result", "type"],
-                                "type_mapping": {
-                                    "target_type": "integer",
-                                    "current_type": "customInteger",
-                                },
+                            "target_type": "string",
+                            "current_type": "singleLineText"
+                        },
+                        {
+                            "target_type": {
+                                "field_type": "array",
+                                "items": "integer"
                             },
+                            "current_type": "formula",
                             "condition": "{{ raw_schema['result']['type'] == 'customInteger' }}",
                         },
                     ],