Update after review

Author: lazebnyi

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1663,8 +1663,8 @@ definitions:
       $parameters:
         type: object
         additionalProperties: true
-  TypesPair:
-    title: Types Pair
+  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.
     type: object
     required:
@@ -1696,37 +1696,33 @@ definitions:
         enum: [SchemaTypeIdentifier]
       schema_pointer:
         title: Schema Path
-        description: List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).
+        description: List of nested fields defining the schema field path to extract. Defaults to [].
         type: array
+        default: []
         items:
           - type: string
         interpolation_content:
           - config
       key_pointer:
         title: Key Path
-        description: List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).
+        description: List of potentially nested fields describing the full path of the field key to extract.
         type: array
         items:
           - type: string
         interpolation_content:
           - config
       type_pointer:
         title: Type Path
-        description: List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).
+        description: List of potentially nested fields describing the full path of the field type to extract.
         type: array
         items:
           - type: string
         interpolation_content:
           - config
-      is_nullable:
-        title: Is Nullable
-        description: Add null to defined field type. This field is automatically set by the CDK.
-        type: boolean
-        default: true
-      types_map:
+      types_mapping:
         type: array
         items:
-          - "$ref": "#/definitions/TypesPair"
+          - "$ref": "#/definitions/TypesMap"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -645,7 +645,7 @@ class HttpResponseFilter(BaseModel):
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
-class TypesPair(BaseModel):
+class TypesMap(BaseModel):
     target_type: Union[str, List[str]]
     current_type: Union[str, List[str]]
 
@@ -654,25 +654,20 @@ class SchemaTypeIdentifier(BaseModel):
     type: Optional[Literal["SchemaTypeIdentifier"]] = None
     schema_pointer: List[str] = Field(
         ...,
-        description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).',
+        description="List of nested fields defining the schema field path to extract. Defaults to [].",
         title="Schema Path",
     )
     key_pointer: List[str] = Field(
         ...,
-        description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).',
+        description="List of potentially nested fields describing the full path of the field key to extract.",
         title="Key Path",
     )
     type_pointer: Optional[List[str]] = Field(
         None,
-        description='List of potentially nested fields describing the full path of the field to extract. Use "*" to extract all values from an array. See more info in the [docs](https://docs.airbyte.com/connector-development/config-based/understanding-the-yaml-file/record-selector).',
+        description="List of potentially nested fields describing the full path of the field type to extract.",
         title="Type Path",
     )
-    is_nullable: Optional[bool] = Field(
-        True,
-        description="Add null to defined field type. This field is automatically set by the CDK.",
-        title="Is Nullable",
-    )
-    types_map: Optional[List[TypesPair]] = None
+    types_mapping: Optional[List[TypesMap]] = None
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py

@@ -61,7 +61,7 @@
     # DynamicSchemaLoader
     "DynamicSchemaLoader.retriever": "SimpleRetriever",
     # SchemaTypeIdentifier
-    "SchemaTypeIdentifier.types_map": "TypesPair",
+    "SchemaTypeIdentifier.types_map": "TypesMap",
 }
 
 # We retain a separate registry for custom components to automatically insert the type if it is missing. This is intended to

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -290,7 +290,7 @@
     SubstreamPartitionRouter as SubstreamPartitionRouterModel,
 )
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    TypesPair as TypesPairModel,
+    TypesMap as TypesMapModel,
 )
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
@@ -357,7 +357,7 @@
     InlineSchemaLoader,
     JsonFileSchemaLoader,
     SchemaTypeIdentifier,
-    TypesPair,
+    TypesMap,
 )
 from airbyte_cdk.sources.declarative.spec import Spec
 from airbyte_cdk.sources.declarative.stream_slicers import StreamSlicer
@@ -454,7 +454,7 @@ def _init_mappings(self) -> None:
             JsonFileSchemaLoaderModel: self.create_json_file_schema_loader,
             DynamicSchemaLoaderModel: self.create_dynamic_schema_loader,
             SchemaTypeIdentifierModel: self.create_schema_type_identifier,
-            TypesPairModel: self.create_types_pair,
+            TypesMapModel: self.create_types_map,
             JwtAuthenticatorModel: self.create_jwt_authenticator,
             LegacyToPerPartitionStateMigrationModel: self.create_legacy_to_per_partition_state_migration,
             ListPartitionRouterModel: self.create_list_partition_router,
@@ -1568,18 +1568,18 @@ def create_inline_schema_loader(
         return InlineSchemaLoader(schema=model.schema_ or {}, parameters={})
 
     @staticmethod
-    def create_types_pair(model: TypesPairModel, **kwargs: Any) -> TypesPair:
-        return TypesPair(target_type=model.target_type, current_type=model.current_type)
+    def create_types_map(model: TypesMapModel, **kwargs: Any) -> TypesMap:
+        return TypesMap(target_type=model.target_type, current_type=model.current_type)
 
     def create_schema_type_identifier(
         self, model: SchemaTypeIdentifierModel, config: Config, **kwargs: Any
     ) -> SchemaTypeIdentifier:
         types_map = []
-        if model.types_map:
+        if model.types_mapping:
             types_map.extend(
                 [
                     self._create_component_from_model(types_pair, config=config)
-                    for types_pair in model.types_map
+                    for types_pair in model.types_mapping
                 ]
             )
         model_schema_pointer: List[Union[InterpolatedString, str]] = [
@@ -1590,14 +1590,11 @@ def create_schema_type_identifier(
             [x for x in model.type_pointer] if model.type_pointer else None
         )
 
-        assert model.is_nullable is not None  # for mypy
-
         return SchemaTypeIdentifier(
             schema_pointer=model_schema_pointer,
             key_pointer=model_key_pointer,
             type_pointer=model_type_pointer,
             types_map=types_map,
-            is_nullable=model.is_nullable,
             parameters=model.parameters or {},
         )
 

airbyte_cdk/sources/declarative/schema/__init__.py

@@ -6,6 +6,6 @@
 from airbyte_cdk.sources.declarative.schema.inline_schema_loader import InlineSchemaLoader
 from airbyte_cdk.sources.declarative.schema.json_file_schema_loader import JsonFileSchemaLoader
 from airbyte_cdk.sources.declarative.schema.schema_loader import SchemaLoader
-from airbyte_cdk.sources.declarative.schema.dynamic_schema_loader import DynamicSchemaLoader, TypesPair, SchemaTypeIdentifier
+from airbyte_cdk.sources.declarative.schema.dynamic_schema_loader import DynamicSchemaLoader, TypesMap, SchemaTypeIdentifier
 
-__all__ = ["JsonFileSchemaLoader", "DefaultSchemaLoader", "SchemaLoader", "InlineSchemaLoader", "DynamicSchemaLoader", "TypesPair", "SchemaTypeIdentifier"]
+__all__ = ["JsonFileSchemaLoader", "DefaultSchemaLoader", "SchemaLoader", "InlineSchemaLoader", "DynamicSchemaLoader", "TypesMap", "SchemaTypeIdentifier"]

airbyte_cdk/sources/declarative/schema/dynamic_schema_loader.py

@@ -16,36 +16,36 @@
 from airbyte_cdk.sources.source import ExperimentalClassWarning
 from airbyte_cdk.sources.types import Config
 
-AIRBYTE_DATA_TYPES = {
-    "string": {"type": "string"},
-    "boolean": {"type": "boolean"},
-    "date": {"type": "string", "format": "date"},
+AIRBYTE_DATA_TYPES: Mapping[str, Mapping[str, Any]] = {
+    "string": {"type": ["null", "string"]},
+    "boolean": {"type": ["null", "boolean"]},
+    "date": {"type": ["null", "string"], "format": "date"},
     "timestamp_without_timezone": {
-        "type": "string",
+        "type": ["null", "string"],
         "format": "date-time",
         "airbyte_type": "timestamp_without_timezone",
     },
-    "timestamp_with_timezone": {"type": "string", "format": "date-time"},
+    "timestamp_with_timezone": {"type": ["null", "string"], "format": "date-time"},
     "time_without_timezone": {
-        "type": "string",
+        "type": ["null", "string"],
         "format": "time",
         "airbyte_type": "time_without_timezone",
     },
     "time_with_timezone": {
-        "type": "string",
+        "type": ["null", "string"],
         "format": "time",
         "airbyte_type": "time_with_timezone",
     },
-    "integer": {"type": "integer"},
-    "number": {"type": "number"},
-    "array": {"type": "array"},
-    "object": {"type": "object"},
+    "integer": {"type": ["null", "integer"]},
+    "number": {"type": ["null", "number"]},
+    "array": {"type": ["null", "array"]},
+    "object": {"type": ["null", "object"]},
 }
 
 
 @deprecated("This class is experimental. Use at your own risk.", category=ExperimentalClassWarning)
 @dataclass(frozen=True)
-class TypesPair:
+class TypesMap:
     """
     Represents a mapping between a current type and its corresponding target type.
     """
@@ -65,8 +65,7 @@ class SchemaTypeIdentifier:
     key_pointer: List[Union[InterpolatedString, str]]
     parameters: InitVar[Mapping[str, Any]]
     type_pointer: Optional[List[Union[InterpolatedString, str]]] = None
-    types_map: Optional[List[TypesPair]] = None
-    is_nullable: bool = True
+    types_map: Optional[List[TypesMap]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.schema_pointer = self._update_pointer(self.schema_pointer, parameters)  # type: ignore[assignment]  # This is reqired field in model
@@ -108,19 +107,24 @@ def get_json_schema(self) -> Mapping[str, Any]:
         Constructs a JSON Schema based on retrieved data.
         """
         properties = {}
-        for retrieved_record in self.retriever.read_records({}):
-            raw_schema = self._extract_data(
+        retrieved_record = next(self.retriever.read_records({}), None)  # type: ignore[call-overload] # read_records return Iterable data type
+
+        raw_schema = (
+            self._extract_data(
                 retrieved_record,  # type: ignore[arg-type] # Expected that retrieved_record will be only Mapping[str, Any]
                 self.schema_type_identifier.schema_pointer,
             )
-            for property_definition in raw_schema:
-                key = self._get_key(property_definition, self.schema_type_identifier.key_pointer)
-                value = self._get_type(
-                    property_definition,
-                    self.schema_type_identifier.type_pointer,
-                    is_nullable=self.schema_type_identifier.is_nullable,
-                )
-                properties[key] = value
+            if retrieved_record
+            else []
+        )
+
+        for property_definition in raw_schema:
+            key = self._get_key(property_definition, self.schema_type_identifier.key_pointer)
+            value = self._get_type(
+                property_definition,
+                self.schema_type_identifier.type_pointer,
+            )
+            properties[key] = value
 
         return {
             "$schema": "http://json-schema.org/draft-07/schema#",
@@ -145,7 +149,6 @@ def _get_type(
         self,
         raw_schema: MutableMapping[str, Any],
         field_type_path: Optional[List[Union[InterpolatedString, str]]],
-        is_nullable: bool = True,
     ) -> Union[Mapping[str, Any], List[Mapping[str, Any]]]:
         """
         Determines the JSON Schema type for a field, supporting nullable and combined types.
@@ -161,15 +164,11 @@ def _get_type(
             and len(mapped_field_type) == 2
             and all(isinstance(item, str) for item in mapped_field_type)
         ):
-            first_type = self._make_field_nullable(
-                self._get_airbyte_type(mapped_field_type[0]), is_nullable
-            )
-            second_type = self._make_field_nullable(
-                self._get_airbyte_type(mapped_field_type[1]), is_nullable
-            )
+            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._make_field_nullable(self._get_airbyte_type(mapped_field_type), is_nullable)
+            return self._get_airbyte_type(mapped_field_type)
         else:
             raise ValueError(
                 f"Invalid data type. Available string or two items list of string. Got {mapped_field_type}."
@@ -187,18 +186,6 @@ def _replace_type_if_not_valid(
                     return types_pair.target_type
         return field_type
 
-    @staticmethod
-    def _make_field_nullable(
-        field_type: Mapping[str, Any], is_nullable: bool = True
-    ) -> Mapping[str, Any]:
-        """
-        Wraps a field type to allow null values if `is_nullable` is True.
-        """
-        updated_field_type = dict(deepcopy(field_type))
-        if is_nullable:
-            updated_field_type["type"] = ["null", updated_field_type["type"]]
-        return updated_field_type
-
     @staticmethod
     def _get_airbyte_type(field_type: str) -> Mapping[str, Any]:
         """
@@ -227,9 +214,4 @@ def _extract_data(
             for path in extraction_path
         ]
 
-        if "*" in path:
-            extracted = dpath.values(body, path)  # type: ignore # extracted will be a MutableMapping, given input data structure
-        else:
-            extracted = dpath.get(body, path, default=default)  # type: ignore # extracted will be a MutableMapping, given input data structure
-
-        return extracted
+        return dpath.get(body, path, default=default)  # type: ignore # extracted will be a MutableMapping, given input data structure