Rollback not related changes

Author: lazebnyi

airbyte_cdk/sources/declarative/extractors/__init__.py

@@ -5,7 +5,6 @@
 from airbyte_cdk.sources.declarative.extractors.combined_extractor import CombinedExtractor
 from airbyte_cdk.sources.declarative.extractors.dpath_extractor import DpathExtractor
 from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector
-from airbyte_cdk.sources.declarative.extractors.key_value_extractor import KeyValueExtractor
 from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
 from airbyte_cdk.sources.declarative.extractors.record_selector import RecordSelector
 from airbyte_cdk.sources.declarative.extractors.response_to_file_extractor import (
@@ -20,6 +19,5 @@
     "RecordFilter",
     "RecordSelector",
     "ResponseToFileExtractor",
-    "KeyValueExtractor",
     "CombinedExtractor",
 ]

airbyte_cdk/sources/declarative/extractors/key_value_extractor.py

@@ -262,7 +262,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
             }
         )
 
-        stream_configs = self._stream_configs(self._source_config) + self.dynamic_streams
+        stream_configs = self._stream_configs(self._source_config) + self._dynamic_stream_configs(self._source_config, config)
 
         api_budget_model = self._source_config.get("api_budget")
         if api_budget_model:

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -85,7 +85,6 @@
 from airbyte_cdk.sources.declarative.extractors import (
     CombinedExtractor,
     DpathExtractor,
-    KeyValueExtractor,
     RecordFilter,
     RecordSelector,
     ResponseToFileExtractor,
@@ -309,9 +308,6 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     KeysToSnakeCase as KeysToSnakeCaseModel,
 )
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
-    KeyValueExtractor as KeyValueExtractorModel,
-)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     LegacySessionTokenAuthenticator as LegacySessionTokenAuthenticatorModel,
 )
@@ -649,7 +645,6 @@ def _init_mappings(self) -> None:
             DefaultErrorHandlerModel: self.create_default_error_handler,
             DefaultPaginatorModel: self.create_default_paginator,
             DpathExtractorModel: self.create_dpath_extractor,
-            KeyValueExtractorModel: self.create_key_value_extractor,
             CombinedExtractorModel: self.create_combined_extractor,
             ResponseToFileExtractorModel: self.create_response_to_file_extractor,
             ExponentialBackoffStrategyModel: self.create_exponential_backoff_strategy,
@@ -2228,22 +2223,6 @@ def create_dpath_extractor(
             parameters=model.parameters or {},
         )
 
-    def create_key_value_extractor(
-        self,
-        model: KeyValueExtractorModel,
-        config: Config,
-        decoder: Optional[Decoder] = JsonDecoder(parameters={}),
-        **kwargs: Any,
-    ) -> KeyValueExtractor:
-        keys_extractor = self._create_component_from_model(
-            model=model.keys_extractor, decoder=decoder, config=config
-        )
-        values_extractor = self._create_component_from_model(
-            model=model.values_extractor, decoder=decoder, config=config
-        )
-
-        return KeyValueExtractor(keys_extractor=keys_extractor, values_extractor=values_extractor)
-
     def create_combined_extractor(
         self,
         model: CombinedExtractorModel,
@@ -2472,14 +2451,10 @@ def create_dynamic_schema_loader(
         schema_type_identifier = self._create_component_from_model(
             model.schema_type_identifier, config=config, parameters=model.parameters or {}
         )
-        schema_filter = self._create_component_from_model(
-            model.schema_filter, config=config, parameters=model.parameters or {}
-        )
         return DynamicSchemaLoader(
             retriever=retriever,
             config=config,
             schema_transformations=schema_transformations,
-            schema_filter=schema_filter,
             schema_type_identifier=schema_type_identifier,
             parameters=model.parameters or {},
         )
@@ -3641,7 +3616,6 @@ def create_components_mapping_definition(
             field_path=field_path,  # type: ignore[arg-type] # field_path can be str and InterpolatedString
             value=interpolated_value,
             value_type=ModelToComponentFactory._json_schema_type_name_to_type(model.value_type),
-            create_or_update=model.create_or_update,
             parameters=model.parameters or {},
         )
 
@@ -3688,24 +3662,16 @@ def create_stream_config(
 
         return StreamConfig(
             configs_pointer=model_configs_pointer,
-            default_values=model.default_values,
             parameters=model.parameters or {},
         )
 
     def create_config_components_resolver(
         self, model: ConfigComponentsResolverModel, config: Config
     ) -> Any:
-        model_stream_configs = (
-            model.stream_config if isinstance(model.stream_config, list) else [model.stream_config]
+        stream_config = self._create_component_from_model(
+            model.stream_config, config=config, parameters=model.parameters or {}
         )
 
-        stream_configs = [
-            self._create_component_from_model(
-                stream_config, config=config, parameters=model.parameters or {}
-            )
-            for stream_config in model_stream_configs
-        ]
-
         components_mapping = [
             self._create_component_from_model(
                 model=components_mapping_definition_model,
@@ -3718,9 +3684,8 @@ def create_config_components_resolver(
         ]
 
         return ConfigComponentsResolver(
-            stream_configs=stream_configs,
+            stream_config=stream_config,
             config=config,
-            components_mapping=components_mapping,
             parameters=model.parameters or {},
         )
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -4,8 +4,7 @@
 
 from copy import deepcopy
 from dataclasses import InitVar, dataclass, field
-from itertools import product
-from typing import Any, Dict, Iterable, List, Mapping, Optional, Union
+from typing import Any, Dict, Iterable, List, Mapping, Union
 
 import dpath
 from typing_extensions import deprecated
@@ -29,7 +28,6 @@ class StreamConfig:
 
     configs_pointer: List[Union[InterpolatedString, str]]
     parameters: InitVar[Mapping[str, Any]]
-    default_values: Optional[List[Any]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.configs_pointer = [
@@ -50,7 +48,7 @@ class ConfigComponentsResolver(ComponentsResolver):
         parameters (InitVar[Mapping[str, Any]]): Additional parameters for interpolation.
     """
 
-    stream_configs: List[StreamConfig]
+    stream_config: StreamConfig
     config: Config
     components_mapping: List[ComponentMappingDefinition]
     parameters: InitVar[Mapping[str, Any]]
@@ -84,7 +82,6 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
                         field_path=field_path,
                         value=interpolated_value,
                         value_type=component_mapping.value_type,
-                        create_or_update=component_mapping.create_or_update,
                         parameters=parameters,
                     )
                 )
@@ -94,35 +91,17 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
                 )
 
     @property
-    def _stream_config(self):
-        def resolve_path(pointer):
-            return [
-                node.eval(self.config) if not isinstance(node, str) else node for node in pointer
-            ]
-
-        def normalize_configs(configs):
-            return configs if isinstance(configs, list) else [configs]
-
-        def prepare_streams():
-            for stream_config in self.stream_configs:
-                path = resolve_path(stream_config.configs_pointer)
-                stream_configs = dpath.get(dict(self.config), path, default=[])
-                stream_configs = normalize_configs(stream_configs)
-                if stream_config.default_values:
-                    stream_configs += stream_config.default_values
-                yield [(i, item) for i, item in enumerate(stream_configs)]
-
-        def merge_combination(combo):
-            result = {}
-            for config_index, (elem_index, elem) in enumerate(combo):
-                if isinstance(elem, dict):
-                    result.update(elem)
-                else:
-                    result.setdefault(f"source_config_{config_index}", (elem_index, elem))
-            return result
-
-        all_indexed_streams = list(prepare_streams())
-        return [merge_combination(combo) for combo in product(*all_indexed_streams)]
+    def _stream_config(self) -> Iterable[Mapping[str, Any]]:
+        path = [
+            node.eval(self.config) if not isinstance(node, str) else node
+            for node in self.stream_config.configs_pointer
+        ]
+        stream_config = dpath.get(dict(self.config), path, default=[])
+
+        if not isinstance(stream_config, list):
+            stream_config = [stream_config]
+
+        return stream_config
 
     def resolve_components(
         self, stream_template_config: Dict[str, Any]
@@ -151,21 +130,7 @@ def resolve_components(
                 )
 
                 path = [path.eval(self.config, **kwargs) for path in resolved_component.field_path]
-                parsed_value = self._parse_yaml_if_possible(value)
-                updated = dpath.set(updated_config, path, parsed_value)
 
-                if parsed_value and not updated and resolved_component.create_or_update:
-                    dpath.new(updated_config, path, parsed_value)
+                dpath.set(updated_config, path, value)
 
             yield updated_config
-
-    @staticmethod
-    def _parse_yaml_if_possible(value: Any) -> Any:
-        if isinstance(value, str):
-            try:
-                import yaml
-
-                return yaml.safe_load(value)
-            except Exception:
-                return value
-        return value

airbyte_cdk/sources/declarative/resolvers/config_components_resolver.py

@@ -10,7 +10,6 @@
 import dpath
 from typing_extensions import deprecated
 
-from airbyte_cdk.sources.declarative.extractors.record_filter import RecordFilter
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever
@@ -127,7 +126,6 @@ class DynamicSchemaLoader(SchemaLoader):
     parameters: InitVar[Mapping[str, Any]]
     schema_type_identifier: SchemaTypeIdentifier
     schema_transformations: List[RecordTransformation] = field(default_factory=lambda: [])
-    schema_filter: Optional[RecordFilter] = None
 
     def get_json_schema(self) -> Mapping[str, Any]:
         """
@@ -153,18 +151,20 @@ def get_json_schema(self) -> Mapping[str, Any]:
             )
             properties[key] = value
 
-        filtred_transformed_properties = self._transform(self._filter(properties))
+        transformed_properties = self._transform(properties, {})
 
         return {
             "$schema": "https://json-schema.org/draft-07/schema#",
             "type": "object",
             "additionalProperties": True,
-            "properties": filtred_transformed_properties,
+            "properties": transformed_properties,
         }
 
     def _transform(
         self,
         properties: Mapping[str, Any],
+        stream_state: StreamState,
+        stream_slice: Optional[StreamSlice] = None,
     ) -> Mapping[str, Any]:
         for transformation in self.schema_transformations:
             transformation.transform(
@@ -173,18 +173,6 @@ def _transform(
             )
         return properties
 
-    def _filter(
-        self,
-        properties: Mapping[str, Any],
-    ) -> Mapping[str, Any]:
-        if self.schema_filter:
-            filtered_properties = {}
-            for property in self.schema_filter.filter_records(properties.items(), {}):
-                filtered_properties[property[0]] = property[1]
-            return filtered_properties
-        else:
-            return properties
-
     def _get_key(
         self,
         raw_schema: MutableMapping[str, Any],