From b87898704dbc286e8b57401108d1e25e34a5a0f8 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Thu, 24 Jul 2025 15:27:22 +0300 Subject: [PATCH 01/21] added ComponentConstructor class --- .../parsers/component_constructor.py | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) create mode 100644 airbyte_cdk/sources/declarative/parsers/component_constructor.py diff --git a/airbyte_cdk/sources/declarative/parsers/component_constructor.py b/airbyte_cdk/sources/declarative/parsers/component_constructor.py new file mode 100644 index 0000000000..b48f287dbb --- /dev/null +++ b/airbyte_cdk/sources/declarative/parsers/component_constructor.py @@ -0,0 +1,72 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +from dataclasses import dataclass +from typing import Any, Callable, Generic, Mapping, Optional, Type, TypeVar + +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType +from airbyte_cdk.sources.types import Config +from pydantic.v1 import BaseModel + +M = TypeVar("M", bound=BaseModel) + + +@dataclass +class ComponentConstructor(Generic[M]): + @classmethod + def resolve_dependencies( + cls, + model: M, + config: Config, + dependency_constructor: Callable[..., Any], + additional_flags: Optional[Mapping[str, Any]] = None, + **kwargs: Any, + ) -> Mapping[str, Any]: + """ + Resolves the component's dependencies, this method should be created in the component, + if there are any dependencies on other components, or we need to adopt / change / adjust / fine-tune + specific component's behavior. + """ + return {} + + @classmethod + def build( + cls, + model: M, + config: Config, + dependency_constructor: Callable[..., Any], + additional_flags: Optional[Mapping[str, Any]], + **kwargs: Any, + ) -> "ComponentConstructor[M]": + """ + Builds up the Component and it's component-specific dependencies. + Order of operations: + - build the dependencies first + - build the component with the resolved dependencies + """ + + # resolve the component dependencies first + resolved_dependencies: Mapping[str, Any] = cls.resolve_dependencies( + model=model, + config=config, + dependency_constructor=dependency_constructor, + additional_flags=additional_flags, + **kwargs, + ) + + # returns the instance of the component class, + # with resolved dependencies and model-specific arguments. + return cls(**resolved_dependencies) + + @staticmethod + def _json_schema_type_name_to_type(value_type: Optional[ValueType]) -> Optional[Type[Any]]: + if not value_type: + return None + names_to_types = { + ValueType.string: str, + ValueType.number: float, + ValueType.integer: int, + ValueType.boolean: bool, + } + return names_to_types[value_type] From 0c27b96406b326a0cc9ff32bb4c9be3899b53821 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Thu, 24 Jul 2025 15:28:28 +0300 Subject: [PATCH 02/21] updated model_to_component.py, set up RecordFilter with resolve_dependencies method --- .../declarative/extractors/record_filter.py | 23 ++++++- .../parsers/model_to_component_factory.py | 61 +++++++++++++------ 2 files changed, 64 insertions(+), 20 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 373669612c..83be23af02 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -2,7 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # from dataclasses import InitVar, dataclass -from typing import Any, Iterable, Mapping, Optional, Union +from typing import Any, Iterable, Mapping, Optional, Union, Callable from airbyte_cdk.sources.declarative.incremental import ( DatetimeBasedCursor, @@ -11,10 +11,14 @@ ) from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + RecordFilter as RecordFilterModel, +) +from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor @dataclass -class RecordFilter: +class RecordFilter(ComponentConstructor[RecordFilterModel]): """ Filter applied on a list of Records @@ -26,6 +30,21 @@ class RecordFilter: config: Config condition: str = "" + @classmethod + def resolve_dependencies( + cls, + model: RecordFilterModel, + config: Config, + dependency_constructor: Callable[..., Any], + additional_flags: Optional[Mapping[str, Any]] = None, + **kwargs: Any, + ) -> Mapping[str, Any]: + return { + "condition": model.condition or "", + "config": config, + "parameters": model.parameters or {}, + } + def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._filter_interpolator = InterpolatedBoolean( condition=self.condition, parameters=parameters diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index c4b9d18f6f..702215cf6a 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -23,6 +23,7 @@ get_args, get_origin, get_type_hints, + TypeVar, ) from isodate import parse_duration @@ -615,6 +616,8 @@ from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction from airbyte_cdk.sources.types import Config from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer +from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor + ComponentDefinition = Mapping[str, Any] @@ -623,8 +626,17 @@ SchemaNormalizationModel.Default: TransformConfig.DefaultSchemaNormalization, } +M = TypeVar("M", bound=BaseModel) +D = TypeVar("D", bound=BaseModel) + class ModelToComponentFactory: + """ + The default Model > Component Factory implementation. + The Custom components are built separately from the default implementations, + to provide the reasonable decoupling from the standard and Custom implementation build technique. + """ + EPOCH_DATETIME_FORMAT = "%s" def __init__( @@ -655,8 +667,19 @@ def __init__( # placeholder for deprecation warnings self._collected_deprecation_logs: List[ConnectorBuilderLogMessage] = [] + # support the dependency constructors with the re-usable parts from this Factory + self._flags = { + "_limit_pages_fetched_per_slice": self._limit_pages_fetched_per_slice, + "_limit_slices_fetched": self._limit_slices_fetched, + "_emit_connector_builder_messages": self._emit_connector_builder_messages, + "_disable_retries": self._disable_retries, + "_message_repository": self._message_repository, + } + def _init_mappings(self) -> None: - self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = { + self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Dict[ + Type[BaseModel], Union[Type[ComponentConstructor], Callable[..., Any]] + ] = { AddedFieldDefinitionModel: self.create_added_field_definition, AddFieldsModel: self.create_add_fields, ApiKeyAuthenticatorModel: self.create_api_key_authenticator, @@ -734,7 +757,7 @@ def _init_mappings(self) -> None: PropertiesFromEndpointModel: self.create_properties_from_endpoint, PropertyChunkingModel: self.create_property_chunking, QueryPropertiesModel: self.create_query_properties, - RecordFilterModel: self.create_record_filter, + RecordFilterModel: RecordFilter, RecordSelectorModel: self.create_record_selector, RemoveFieldsModel: self.create_remove_fields, RequestPathModel: self.create_request_path, @@ -803,20 +826,30 @@ def create_component( model=declarative_component_model, config=config, **kwargs ) - def _create_component_from_model(self, model: BaseModel, config: Config, **kwargs: Any) -> Any: + def _create_component_from_model( + self, model: BaseModel, config: Config, **kwargs: Any + ) -> ComponentConstructor[BaseModel]: if model.__class__ not in self.PYDANTIC_MODEL_TO_CONSTRUCTOR: raise ValueError( f"{model.__class__} with attributes {model} is not a valid component type" ) - component_constructor = self.PYDANTIC_MODEL_TO_CONSTRUCTOR.get(model.__class__) - if not component_constructor: - raise ValueError(f"Could not find constructor for {model.__class__}") - # collect deprecation warnings for supported models. - if isinstance(model, BaseModelWithDeprecations): - self._collect_model_deprecations(model) + component = self.PYDANTIC_MODEL_TO_CONSTRUCTOR.get(model.__class__) + if not component: + raise ValueError(f"Could not find constructor for {model.__class__}") - return component_constructor(model=model, config=config, **kwargs) + if inspect.isclass(component) and issubclass(component, ComponentConstructor): + # Default components flow + component_instance: ComponentConstructor[BaseModel] = component.build( + model=model, + config=config, + dependency_constructor=self._create_component_from_model, + additional_flags=self._flags, + **kwargs, + ) + return component_instance + else: + return component(model=model, config=config, **kwargs) def get_model_deprecations(self) -> List[ConnectorBuilderLogMessage]: """ @@ -2999,14 +3032,6 @@ def create_query_properties( parameters=model.parameters or {}, ) - @staticmethod - def create_record_filter( - model: RecordFilterModel, config: Config, **kwargs: Any - ) -> RecordFilter: - return RecordFilter( - condition=model.condition or "", config=config, parameters=model.parameters or {} - ) - @staticmethod def create_request_path(model: RequestPathModel, config: Config, **kwargs: Any) -> RequestPath: return RequestPath(parameters={}) From 2d8129b2893ccddc18f49e7b2622cb4982430cd7 Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Thu, 24 Jul 2025 12:33:02 +0000 Subject: [PATCH 03/21] Auto-fix lint and format issues --- airbyte_cdk/sources/declarative/extractors/record_filter.py | 4 ++-- .../sources/declarative/parsers/component_constructor.py | 3 ++- .../declarative/parsers/model_to_component_factory.py | 5 ++--- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 83be23af02..4f44278242 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -2,7 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # from dataclasses import InitVar, dataclass -from typing import Any, Iterable, Mapping, Optional, Union, Callable +from typing import Any, Callable, Iterable, Mapping, Optional, Union from airbyte_cdk.sources.declarative.incremental import ( DatetimeBasedCursor, @@ -10,11 +10,11 @@ PerPartitionWithGlobalCursor, ) from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean -from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( RecordFilter as RecordFilterModel, ) from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor +from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState @dataclass diff --git a/airbyte_cdk/sources/declarative/parsers/component_constructor.py b/airbyte_cdk/sources/declarative/parsers/component_constructor.py index b48f287dbb..8c9442df23 100644 --- a/airbyte_cdk/sources/declarative/parsers/component_constructor.py +++ b/airbyte_cdk/sources/declarative/parsers/component_constructor.py @@ -5,9 +5,10 @@ from dataclasses import dataclass from typing import Any, Callable, Generic, Mapping, Optional, Type, TypeVar +from pydantic.v1 import BaseModel + from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType from airbyte_cdk.sources.types import Config -from pydantic.v1 import BaseModel M = TypeVar("M", bound=BaseModel) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 702215cf6a..7a3f76af2a 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -18,12 +18,12 @@ MutableMapping, Optional, Type, + TypeVar, Union, cast, get_args, get_origin, get_type_hints, - TypeVar, ) from isodate import parse_duration @@ -447,6 +447,7 @@ from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ZipfileDecoder as ZipfileDecoderModel, ) +from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import ( COMPONENTS_MODULE_NAME, SDM_COMPONENTS_MODULE_NAME, @@ -616,8 +617,6 @@ from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction from airbyte_cdk.sources.types import Config from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer -from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor - ComponentDefinition = Mapping[str, Any] From e1a4e7eaf41df90ac3d5f037754b210cf514019e Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Thu, 24 Jul 2025 16:32:52 +0300 Subject: [PATCH 04/21] fix mypy --- .../parsers/model_to_component_factory.py | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 7a3f76af2a..5692d7e349 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -677,7 +677,7 @@ def __init__( def _init_mappings(self) -> None: self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Dict[ - Type[BaseModel], Union[Type[ComponentConstructor], Callable[..., Any]] + Type[BaseModel], Union[Type[ComponentConstructor[Any]], Callable[..., Any]] ] = { AddedFieldDefinitionModel: self.create_added_field_definition, AddFieldsModel: self.create_add_fields, @@ -826,8 +826,11 @@ def create_component( ) def _create_component_from_model( - self, model: BaseModel, config: Config, **kwargs: Any - ) -> ComponentConstructor[BaseModel]: + self, + model: BaseModel, + config: Config, + **kwargs: Any, + ) -> Any: # TODO: change -> Any to -> ComponentConstructor[BaseModel] once all components are updated with ComponentConstructor logic if model.__class__ not in self.PYDANTIC_MODEL_TO_CONSTRUCTOR: raise ValueError( f"{model.__class__} with attributes {model} is not a valid component type" @@ -847,8 +850,12 @@ def _create_component_from_model( **kwargs, ) return component_instance - else: + if inspect.ismethod(component): return component(model=model, config=config, **kwargs) + raise ValueError( + f"Unexpected component mapping type for {model.__class__}. " + f"Instance shource be one of ComponentConstructor or method implemented in ModelToComponentFactory" + ) def get_model_deprecations(self) -> List[ConnectorBuilderLogMessage]: """ From 2b569f5a24efffbba7b570d44ebe32cdf713e690 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Thu, 24 Jul 2025 16:41:50 +0300 Subject: [PATCH 05/21] added check for function --- .../sources/declarative/parsers/model_to_component_factory.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 5692d7e349..858e26311d 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -850,7 +850,7 @@ def _create_component_from_model( **kwargs, ) return component_instance - if inspect.ismethod(component): + if inspect.ismethod(component) or inspect.isfunction(component): return component(model=model, config=config, **kwargs) raise ValueError( f"Unexpected component mapping type for {model.__class__}. " From bd551fca0c1b7f1e19c18413bc6ce3e4adc6beb8 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 12:28:51 +0300 Subject: [PATCH 06/21] fix typo in error message --- .../sources/declarative/parsers/model_to_component_factory.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 858e26311d..514a2ec247 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -763,7 +763,7 @@ def _init_mappings(self) -> None: RequestOptionModel: self.create_request_option, LegacySessionTokenAuthenticatorModel: self.create_legacy_session_token_authenticator, SelectiveAuthenticatorModel: self.create_selective_authenticator, - SimpleRetrieverModel: self.create_simple_retriever, + SimpleRetrieverModel: SimpleRetriever, StateDelegatingStreamModel: self.create_state_delegating_stream, SpecModel: self.create_spec, SubstreamPartitionRouterModel: self.create_substream_partition_router, @@ -854,7 +854,7 @@ def _create_component_from_model( return component(model=model, config=config, **kwargs) raise ValueError( f"Unexpected component mapping type for {model.__class__}. " - f"Instance shource be one of ComponentConstructor or method implemented in ModelToComponentFactory" + f"Instance should be one of ComponentConstructor or method implemented in ModelToComponentFactory" ) def get_model_deprecations(self) -> List[ConnectorBuilderLogMessage]: From 01d2f876b8b273d76712616837bd4ea1d39d0fb5 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 16:15:59 +0300 Subject: [PATCH 07/21] added AdditionalFlags to ComponentConstructor --- .../parsers/component_constructor.py | 34 +++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/component_constructor.py b/airbyte_cdk/sources/declarative/parsers/component_constructor.py index 8c9442df23..7ee9c3546b 100644 --- a/airbyte_cdk/sources/declarative/parsers/component_constructor.py +++ b/airbyte_cdk/sources/declarative/parsers/component_constructor.py @@ -7,12 +7,42 @@ from pydantic.v1 import BaseModel +from airbyte_cdk.sources.message import MessageRepository from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType from airbyte_cdk.sources.types import Config +from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager + M = TypeVar("M", bound=BaseModel) +@dataclass +class AdditionalFlags: + def __init__( + self, + emit_connector_builder_messages: bool, + disable_retries: bool, + message_repository: MessageRepository, + connector_state_manager: ConnectorStateManager, + limit_pages_fetched_per_slice: Optional[int], + limit_slices_fetched: Optional[int], + ): + self.emit_connector_builder_messages = emit_connector_builder_messages + self.disable_retries = disable_retries + self.message_repository = message_repository + self.connector_state_manager = connector_state_manager + self.limit_pages_fetched_per_slice = limit_pages_fetched_per_slice + self.limit_slices_fetched = limit_slices_fetched + + @property + def should_limit_slices_fetched(self) -> bool: + """ + Returns True if the number of slices fetched should be limited, False otherwise. + This is used to limit the number of slices fetched during tests. + """ + return bool(self.limit_slices_fetched or self.emit_connector_builder_messages) + + @dataclass class ComponentConstructor(Generic[M]): @classmethod @@ -21,7 +51,7 @@ def resolve_dependencies( model: M, config: Config, dependency_constructor: Callable[..., Any], - additional_flags: Optional[Mapping[str, Any]] = None, + additional_flags: AdditionalFlags, **kwargs: Any, ) -> Mapping[str, Any]: """ @@ -37,7 +67,7 @@ def build( model: M, config: Config, dependency_constructor: Callable[..., Any], - additional_flags: Optional[Mapping[str, Any]], + additional_flags: AdditionalFlags, **kwargs: Any, ) -> "ComponentConstructor[M]": """ From fafab4db7a72ee9b58347e601b63eaa0f9b4c6f4 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 16:18:04 +0300 Subject: [PATCH 08/21] added ComponentConstructor to QueryProperties --- .../query_properties/query_properties.py | 40 ++++++++++++++++++- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py b/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py index 37b26d1713..e91f831e0c 100644 --- a/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py +++ b/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py @@ -1,17 +1,24 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. from dataclasses import InitVar, dataclass -from typing import Any, Iterable, List, Mapping, Optional, Union +from typing import Any, Iterable, List, Mapping, Optional, Union, Callable from airbyte_cdk.sources.declarative.requesters.query_properties import ( PropertiesFromEndpoint, PropertyChunking, ) from airbyte_cdk.sources.types import Config, StreamSlice +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + QueryProperties as QueryPropertiesModel, +) +from airbyte_cdk.sources.declarative.parsers.component_constructor import ( + ComponentConstructor, + AdditionalFlags, +) @dataclass -class QueryProperties: +class QueryProperties(ComponentConstructor[QueryPropertiesModel]): """ Low-code component that encompasses the behavior to inject additional property values into the outbound API requests. Property values can be defined statically within the manifest or dynamically by making requests @@ -25,6 +32,35 @@ class QueryProperties: config: Config parameters: InitVar[Mapping[str, Any]] + @classmethod + def resolve_dependencies( + cls, + model: QueryPropertiesModel, + config: Config, + dependency_constructor: Callable[..., Any], + additional_flags: AdditionalFlags, + **kwargs: Any, + ) -> Mapping[str, Any]: + if isinstance(model.property_list, list): + property_list = model.property_list + else: + property_list = dependency_constructor( + model=model.property_list, config=config, **kwargs + ) + + property_chunking = ( + dependency_constructor(model=model.property_chunking, config=config, **kwargs) + if model.property_chunking + else None + ) + return { + "property_list": property_list, + "always_include_properties": model.always_include_properties, + "property_chunking": property_chunking, + "config": config, + "parameters": model.parameters or {}, + } + def get_request_property_chunks( self, stream_slice: Optional[StreamSlice] = None ) -> Iterable[List[str]]: From 57706a733ee6bc3bff298b5dfafc541f1e3f1a98 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 16:19:02 +0300 Subject: [PATCH 09/21] updated RecordFilter --- .../sources/declarative/extractors/record_filter.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 4f44278242..809a190bcd 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -13,7 +13,10 @@ from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( RecordFilter as RecordFilterModel, ) -from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor +from airbyte_cdk.sources.declarative.parsers.component_constructor import ( + ComponentConstructor, + AdditionalFlags, +) from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState @@ -36,7 +39,7 @@ def resolve_dependencies( model: RecordFilterModel, config: Config, dependency_constructor: Callable[..., Any], - additional_flags: Optional[Mapping[str, Any]] = None, + additional_flags: AdditionalFlags, **kwargs: Any, ) -> Mapping[str, Any]: return { From bc4a36f232e3462cff2fbba9ab69632fb788e79e Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 16:20:05 +0300 Subject: [PATCH 10/21] updated ModelToComponent --- .../parsers/model_to_component_factory.py | 292 +----------------- 1 file changed, 13 insertions(+), 279 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 514a2ec247..02f262ae5c 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -447,7 +447,10 @@ from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ZipfileDecoder as ZipfileDecoderModel, ) -from airbyte_cdk.sources.declarative.parsers.component_constructor import ComponentConstructor +from airbyte_cdk.sources.declarative.parsers.component_constructor import ( + ComponentConstructor, + AdditionalFlags, +) from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import ( COMPONENTS_MODULE_NAME, SDM_COMPONENTS_MODULE_NAME, @@ -667,13 +670,14 @@ def __init__( self._collected_deprecation_logs: List[ConnectorBuilderLogMessage] = [] # support the dependency constructors with the re-usable parts from this Factory - self._flags = { - "_limit_pages_fetched_per_slice": self._limit_pages_fetched_per_slice, - "_limit_slices_fetched": self._limit_slices_fetched, - "_emit_connector_builder_messages": self._emit_connector_builder_messages, - "_disable_retries": self._disable_retries, - "_message_repository": self._message_repository, - } + self._flags = AdditionalFlags( + emit_connector_builder_messages=self._emit_connector_builder_messages, + disable_retries=self._disable_retries, + message_repository=self._message_repository, + connector_state_manager=self._connector_state_manager, + limit_pages_fetched_per_slice=self._limit_pages_fetched_per_slice, + limit_slices_fetched=self._limit_slices_fetched, + ) def _init_mappings(self) -> None: self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Dict[ @@ -755,7 +759,7 @@ def _init_mappings(self) -> None: PredicateValidatorModel: self.create_predicate_validator, PropertiesFromEndpointModel: self.create_properties_from_endpoint, PropertyChunkingModel: self.create_property_chunking, - QueryPropertiesModel: self.create_query_properties, + QueryPropertiesModel: QueryProperties, RecordFilterModel: RecordFilter, RecordSelectorModel: self.create_record_selector, RemoveFieldsModel: self.create_remove_fields, @@ -3012,32 +3016,6 @@ def create_property_chunking( parameters=model.parameters or {}, ) - def create_query_properties( - self, model: QueryPropertiesModel, config: Config, **kwargs: Any - ) -> QueryProperties: - if isinstance(model.property_list, list): - property_list = model.property_list - else: - property_list = self._create_component_from_model( - model=model.property_list, config=config, **kwargs - ) - - property_chunking = ( - self._create_component_from_model( - model=model.property_chunking, config=config, **kwargs - ) - if model.property_chunking - else None - ) - - return QueryProperties( - property_list=property_list, - always_include_properties=model.always_include_properties, - property_chunking=property_chunking, - config=config, - parameters=model.parameters or {}, - ) - @staticmethod def create_request_path(model: RequestPathModel, config: Config, **kwargs: Any) -> RequestPath: return RequestPath(parameters={}) @@ -3168,237 +3146,6 @@ def create_legacy_session_token_authenticator( parameters=model.parameters or {}, ) - def create_simple_retriever( - self, - model: SimpleRetrieverModel, - config: Config, - *, - name: str, - primary_key: Optional[Union[str, List[str], List[List[str]]]], - stream_slicer: Optional[StreamSlicer], - request_options_provider: Optional[RequestOptionsProvider] = None, - stop_condition_on_cursor: bool = False, - client_side_incremental_sync: Optional[Dict[str, Any]] = None, - transformations: List[RecordTransformation], - file_uploader: Optional[DefaultFileUploader] = None, - incremental_sync: Optional[ - Union[ - IncrementingCountCursorModel, DatetimeBasedCursorModel, CustomIncrementalSyncModel - ] - ] = None, - use_cache: Optional[bool] = None, - log_formatter: Optional[Callable[[Response], Any]] = None, - **kwargs: Any, - ) -> SimpleRetriever: - def _get_url() -> str: - """ - Closure to get the URL from the requester. This is used to get the URL in the case of a lazy retriever. - This is needed because the URL is not set until the requester is created. - """ - - _url: str = ( - model.requester.url - if hasattr(model.requester, "url") and model.requester.url is not None - else requester.get_url() - ) - _url_base: str = ( - model.requester.url_base - if hasattr(model.requester, "url_base") and model.requester.url_base is not None - else requester.get_url_base() - ) - - return _url or _url_base - - decoder = ( - self._create_component_from_model(model=model.decoder, config=config) - if model.decoder - else JsonDecoder(parameters={}) - ) - record_selector = self._create_component_from_model( - model=model.record_selector, - name=name, - config=config, - decoder=decoder, - transformations=transformations, - client_side_incremental_sync=client_side_incremental_sync, - file_uploader=file_uploader, - ) - - query_properties: Optional[QueryProperties] = None - query_properties_key: Optional[str] = None - if self._query_properties_in_request_parameters(model.requester): - # It is better to be explicit about an error if PropertiesFromEndpoint is defined in multiple - # places instead of default to request_parameters which isn't clearly documented - if ( - hasattr(model.requester, "fetch_properties_from_endpoint") - and model.requester.fetch_properties_from_endpoint - ): - raise ValueError( - f"PropertiesFromEndpoint should only be specified once per stream, but found in {model.requester.type}.fetch_properties_from_endpoint and {model.requester.type}.request_parameters" - ) - - query_properties_definitions = [] - for key, request_parameter in model.requester.request_parameters.items(): # type: ignore # request_parameters is already validated to be a Mapping using _query_properties_in_request_parameters() - if isinstance(request_parameter, QueryPropertiesModel): - query_properties_key = key - query_properties_definitions.append(request_parameter) - - if len(query_properties_definitions) > 1: - raise ValueError( - f"request_parameters only supports defining one QueryProperties field, but found {len(query_properties_definitions)} usages" - ) - - if len(query_properties_definitions) == 1: - query_properties = self._create_component_from_model( - model=query_properties_definitions[0], config=config - ) - elif ( - hasattr(model.requester, "fetch_properties_from_endpoint") - and model.requester.fetch_properties_from_endpoint - ): - # todo: Deprecate this condition once dependent connectors migrate to query_properties - query_properties_definition = QueryPropertiesModel( - type="QueryProperties", - property_list=model.requester.fetch_properties_from_endpoint, - always_include_properties=None, - property_chunking=None, - ) # type: ignore # $parameters has a default value - - query_properties = self.create_query_properties( - model=query_properties_definition, - config=config, - ) - elif hasattr(model.requester, "query_properties") and model.requester.query_properties: - query_properties = self.create_query_properties( - model=model.requester.query_properties, - config=config, - ) - - requester = self._create_component_from_model( - model=model.requester, - decoder=decoder, - name=name, - query_properties_key=query_properties_key, - use_cache=use_cache, - config=config, - ) - - # Define cursor only if per partition or common incremental support is needed - cursor = stream_slicer if isinstance(stream_slicer, DeclarativeCursor) else None - - if ( - not isinstance(stream_slicer, DatetimeBasedCursor) - or type(stream_slicer) is not DatetimeBasedCursor - ): - # Many of the custom component implementations of DatetimeBasedCursor override get_request_params() (or other methods). - # Because we're decoupling RequestOptionsProvider from the Cursor, custom components will eventually need to reimplement - # their own RequestOptionsProvider. However, right now the existing StreamSlicer/Cursor still can act as the SimpleRetriever's - # request_options_provider - request_options_provider = stream_slicer or DefaultRequestOptionsProvider(parameters={}) - elif not request_options_provider: - request_options_provider = DefaultRequestOptionsProvider(parameters={}) - - stream_slicer = stream_slicer or SinglePartitionRouter(parameters={}) - if self._should_limit_slices_fetched(): - stream_slicer = cast( - StreamSlicer, - StreamSlicerTestReadDecorator( - wrapped_slicer=stream_slicer, - maximum_number_of_slices=self._limit_slices_fetched or 5, - ), - ) - - cursor_used_for_stop_condition = cursor if stop_condition_on_cursor else None - paginator = ( - self._create_component_from_model( - model=model.paginator, - config=config, - url_base=_get_url(), - extractor_model=model.record_selector.extractor, - decoder=decoder, - cursor_used_for_stop_condition=cursor_used_for_stop_condition, - ) - if model.paginator - else NoPagination(parameters={}) - ) - - ignore_stream_slicer_parameters_on_paginated_requests = ( - model.ignore_stream_slicer_parameters_on_paginated_requests or False - ) - - if ( - model.partition_router - and isinstance(model.partition_router, SubstreamPartitionRouterModel) - and not bool(self._connector_state_manager.get_stream_state(name, None)) - and any( - parent_stream_config.lazy_read_pointer - for parent_stream_config in model.partition_router.parent_stream_configs - ) - ): - if incremental_sync: - if incremental_sync.type != "DatetimeBasedCursor": - raise ValueError( - f"LazySimpleRetriever only supports DatetimeBasedCursor. Found: {incremental_sync.type}." - ) - - elif incremental_sync.step or incremental_sync.cursor_granularity: - raise ValueError( - f"Found more that one slice per parent. LazySimpleRetriever only supports single slice read for stream - {name}." - ) - - if model.decoder and model.decoder.type != "JsonDecoder": - raise ValueError( - f"LazySimpleRetriever only supports JsonDecoder. Found: {model.decoder.type}." - ) - - return LazySimpleRetriever( - name=name, - paginator=paginator, - primary_key=primary_key, - requester=requester, - record_selector=record_selector, - stream_slicer=stream_slicer, - request_option_provider=request_options_provider, - cursor=cursor, - config=config, - ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests, - parameters=model.parameters or {}, - ) - - return SimpleRetriever( - name=name, - paginator=paginator, - primary_key=primary_key, - requester=requester, - record_selector=record_selector, - stream_slicer=stream_slicer, - request_option_provider=request_options_provider, - cursor=cursor, - config=config, - ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests, - additional_query_properties=query_properties, - log_formatter=self._get_log_formatter(log_formatter, name), - parameters=model.parameters or {}, - ) - - def _get_log_formatter( - self, log_formatter: Callable[[Response], Any] | None, name: str - ) -> Callable[[Response], Any] | None: - if self._should_limit_slices_fetched(): - return ( - ( - lambda response: format_http_message( - response, - f"Stream '{name}' request", - f"Request performed in order to extract records for stream '{name}'", - name, - ) - ) - if not log_formatter - else log_formatter - ) - return None - def _should_limit_slices_fetched(self) -> bool: """ Returns True if the number of slices fetched should be limited, False otherwise. @@ -3406,19 +3153,6 @@ def _should_limit_slices_fetched(self) -> bool: """ return bool(self._limit_slices_fetched or self._emit_connector_builder_messages) - @staticmethod - def _query_properties_in_request_parameters( - requester: Union[HttpRequesterModel, CustomRequesterModel], - ) -> bool: - if not hasattr(requester, "request_parameters"): - return False - request_parameters = requester.request_parameters - if request_parameters and isinstance(request_parameters, Mapping): - for request_parameter in request_parameters.values(): - if isinstance(request_parameter, QueryPropertiesModel): - return True - return False - @staticmethod def _remove_query_properties( request_parameters: Mapping[str, Union[str, QueryPropertiesModel]], From ba33479f3bbd7cea153a36b1fb09a01db8d4c68d Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 16:20:46 +0300 Subject: [PATCH 11/21] updated SimpleRetriever with ComponentConstructor --- .../retrievers/simple_retriever.py | 339 +++++++++++++++++- 1 file changed, 338 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 4ee2422cd5..cf8cb0a3ee 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -18,6 +18,8 @@ Set, Tuple, Union, + Dict, + cast, ) import requests @@ -46,12 +48,49 @@ from airbyte_cdk.sources.streams.core import StreamData from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState from airbyte_cdk.utils.mapping_helpers import combine_mappings +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + SimpleRetriever as SimpleRetrieverModel, +) +from airbyte_cdk.sources.declarative.parsers.component_constructor import ( + ComponentConstructor, + AdditionalFlags, +) +from airbyte_cdk.sources.declarative.transformations import RecordTransformation +from airbyte_cdk.sources.declarative.decoders import JsonDecoder +from pydantic import BaseModel +from airbyte_cdk.sources.declarative.retrievers.file_uploader import DefaultFileUploader +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + IncrementingCountCursor as IncrementingCountCursorModel, +) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + DatetimeBasedCursor as DatetimeBasedCursorModel, +) + +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + CustomIncrementalSync as CustomIncrementalSyncModel, +) +from requests import Response +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + HttpRequester as HttpRequesterModel, +) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + CustomRequester as CustomRequesterModel, +) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + QueryProperties as QueryPropertiesModel, +) +from airbyte_cdk.sources.declarative.requesters.query_properties import QueryProperties +from airbyte_cdk.sources.declarative.incremental import DatetimeBasedCursor +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + SubstreamPartitionRouter as SubstreamPartitionRouterModel, +) +from airbyte_cdk.sources.declarative.stream_slicers import StreamSlicerTestReadDecorator FULL_REFRESH_SYNC_COMPLETE_KEY = "__ab_full_refresh_sync_complete" @dataclass -class SimpleRetriever(Retriever): +class SimpleRetriever(Retriever, ComponentConstructor[SimpleRetrieverModel]): """ Retrieves records by synchronously sending requests to fetch records. @@ -93,6 +132,304 @@ class SimpleRetriever(Retriever): ignore_stream_slicer_parameters_on_paginated_requests: bool = False additional_query_properties: Optional[QueryProperties] = None log_formatter: Optional[Callable[[requests.Response], Any]] = None + _should_use_lazy_simple_retriever: bool = False + + @classmethod + def _validate_if_lazy_simple_retriever_should_be_applied( + cls, + name: str, + model: SimpleRetrieverModel, + additional_flags: AdditionalFlags, + incremental_sync: Optional[ + Union[ + IncrementingCountCursorModel, DatetimeBasedCursorModel, CustomIncrementalSyncModel + ] + ] = None, + ) -> bool: + if ( + model.partition_router + and isinstance(model.partition_router, SubstreamPartitionRouterModel) + and not bool(additional_flags.connector_state_manager.get_stream_state(name, None)) + and any( + parent_stream_config.lazy_read_pointer + for parent_stream_config in model.partition_router.parent_stream_configs + ) + ): + if incremental_sync: + if incremental_sync.type != "DatetimeBasedCursor": + raise ValueError( + f"LazySimpleRetriever only supports DatetimeBasedCursor. Found: {incremental_sync.type}." + ) + + elif incremental_sync.step or incremental_sync.cursor_granularity: + raise ValueError( + f"Found more that one slice per parent. LazySimpleRetriever only supports single slice read for stream - {name}." + ) + + if model.decoder and model.decoder.type != "JsonDecoder": + raise ValueError( + f"LazySimpleRetriever only supports JsonDecoder. Found: {model.decoder.type}." + ) + cls._should_use_lazy_simple_retriever = True + return True + + return False + + @classmethod + def resolve_dependencies( + cls, + model: SimpleRetrieverModel, + config: Config, + dependency_constructor: Callable[..., Any], + additional_flags: AdditionalFlags, + *, + name: Optional[str] = None, + primary_key: Optional[Union[str, List[str], List[List[str]]]] = None, + stream_slicer: Optional[StreamSlicer] = None, + request_options_provider: Optional[RequestOptionsProvider] = None, + stop_condition_on_cursor: bool = False, + client_side_incremental_sync: Optional[Dict[str, Any]] = None, + transformations: Optional[List[RecordTransformation]] = None, + file_uploader: Optional[DefaultFileUploader] = None, + incremental_sync: Optional[ + Union[ + IncrementingCountCursorModel, DatetimeBasedCursorModel, CustomIncrementalSyncModel + ] + ] = None, + use_cache: Optional[bool] = None, + log_formatter: Optional[Callable[[Response], Any]] = None, + **kwargs: Any, + ) -> Mapping[str, Any]: + if not name: + raise ValueError(f"name argument is required to instance a {cls.__name__}") + + def _get_url() -> str: + """ + Closure to get the URL from the requester. This is used to get the URL in the case of a lazy retriever. + This is needed because the URL is not set until the requester is created. + """ + + _url: str = ( + model.requester.url + if hasattr(model.requester, "url") and model.requester.url is not None + else requester.get_url() + ) + _url_base: str = ( + model.requester.url_base + if hasattr(model.requester, "url_base") and model.requester.url_base is not None + else requester.get_url_base() + ) + + return _url or _url_base + + def query_properties_in_request_parameters( + requester: Union[HttpRequesterModel, CustomRequesterModel], + ) -> bool: + if not hasattr(requester, "request_parameters"): + return False + request_parameters = requester.request_parameters + if request_parameters and isinstance(request_parameters, Mapping): + for request_parameter in request_parameters.values(): + if isinstance(request_parameter, QueryPropertiesModel): + return True + return False + + def _get_log_formatter( + log_formatter: Callable[[Response], Any] | None, + name: str, + additional_flags: AdditionalFlags, + ) -> Callable[[Response], Any] | None: + if additional_flags.should_limit_slices_fetched: + return ( + ( + lambda response: format_http_message( + response, + f"Stream '{name}' request", + f"Request performed in order to extract records for stream '{name}'", + name, + ) + ) + if not log_formatter + else log_formatter + ) + return None + + decoder = ( + dependency_constructor(model=model.decoder, config=config) + if model.decoder + else JsonDecoder(parameters={}) + ) + record_selector = dependency_constructor( + model=model.record_selector, + name=name, + config=config, + decoder=decoder, + transformations=transformations, + client_side_incremental_sync=client_side_incremental_sync, + file_uploader=file_uploader, + ) + + query_properties: Optional[QueryProperties] = None + query_properties_key: Optional[str] = None + if query_properties_in_request_parameters(model.requester): + # It is better to be explicit about an error if PropertiesFromEndpoint is defined in multiple + # places instead of default to request_parameters which isn't clearly documented + if ( + hasattr(model.requester, "fetch_properties_from_endpoint") + and model.requester.fetch_properties_from_endpoint + ): + raise ValueError( + f"PropertiesFromEndpoint should only be specified once per stream, but found in {model.requester.type}.fetch_properties_from_endpoint and {model.requester.type}.request_parameters" + ) + + query_properties_definitions = [] + for key, request_parameter in model.requester.request_parameters.items(): # type: ignore # request_parameters is already validated to be a Mapping using query_properties_in_request_parameters() + if isinstance(request_parameter, QueryPropertiesModel): + query_properties_key = key + query_properties_definitions.append(request_parameter) + + if len(query_properties_definitions) > 1: + raise ValueError( + f"request_parameters only supports defining one QueryProperties field, but found {len(query_properties_definitions)} usages" + ) + + if len(query_properties_definitions) == 1: + query_properties = dependency_constructor( + model=query_properties_definitions[0], config=config + ) + elif ( + hasattr(model.requester, "fetch_properties_from_endpoint") + and model.requester.fetch_properties_from_endpoint + ): + # todo: Deprecate this condition once dependent connectors migrate to query_properties + query_properties_definition = QueryPropertiesModel( + type="QueryProperties", + property_list=model.requester.fetch_properties_from_endpoint, + always_include_properties=None, + property_chunking=None, + ) # type: ignore # $parameters has a default value + + query_properties = dependency_constructor( + model=query_properties_definition, + config=config, + dependency_constructor=dependency_constructor, + additional_flags=additional_flags, + **kwargs, + ) + + elif hasattr(model.requester, "query_properties") and model.requester.query_properties: + query_properties = dependency_constructor( + model=model.requester.query_properties, + config=config, + dependency_constructor=dependency_constructor, + additional_flags=additional_flags, + **kwargs, + ) + + requester = dependency_constructor( + model=model.requester, + decoder=decoder, + name=name, + query_properties_key=query_properties_key, + use_cache=use_cache, + config=config, + ) + + # Define cursor only if per partition or common incremental support is needed + cursor = stream_slicer if isinstance(stream_slicer, DeclarativeCursor) else None + + if ( + not isinstance(stream_slicer, DatetimeBasedCursor) + or type(stream_slicer) is not DatetimeBasedCursor + ): + # Many of the custom component implementations of DatetimeBasedCursor override get_request_params() (or other methods). + # Because we're decoupling RequestOptionsProvider from the Cursor, custom components will eventually need to reimplement + # their own RequestOptionsProvider. However, right now the existing StreamSlicer/Cursor still can act as the SimpleRetriever's + # request_options_provider + request_options_provider = stream_slicer or DefaultRequestOptionsProvider(parameters={}) + elif not request_options_provider: + request_options_provider = DefaultRequestOptionsProvider(parameters={}) + + stream_slicer = stream_slicer or SinglePartitionRouter(parameters={}) + if additional_flags.should_limit_slices_fetched: + stream_slicer = cast( + StreamSlicer, + StreamSlicerTestReadDecorator( + wrapped_slicer=stream_slicer, + maximum_number_of_slices=additional_flags.limit_slices_fetched or 5, + ), + ) + + cursor_used_for_stop_condition = cursor if stop_condition_on_cursor else None + paginator = ( + dependency_constructor( + model=model.paginator, + config=config, + url_base=_get_url(), + extractor_model=model.record_selector.extractor, + decoder=decoder, + cursor_used_for_stop_condition=cursor_used_for_stop_condition, + ) + if model.paginator + else NoPagination(parameters={}) + ) + + ignore_stream_slicer_parameters_on_paginated_requests = ( + model.ignore_stream_slicer_parameters_on_paginated_requests or False + ) + + resolved_dependencies = { + "name": name, + "paginator": paginator, + "primary_key": primary_key, + "requester": requester, + "record_selector": record_selector, + "stream_slicer": stream_slicer, + "request_option_provider": request_options_provider, + "cursor": cursor, + "config": config, + "ignore_stream_slicer_parameters_on_paginated_requests": ignore_stream_slicer_parameters_on_paginated_requests, + "parameters": model.parameters or {}, + } + + if cls._validate_if_lazy_simple_retriever_should_be_applied( + name, model, additional_flags, incremental_sync + ): + return resolved_dependencies + + resolved_dependencies.update( + { + "additional_query_properties": query_properties, + "log_formatter": _get_log_formatter(log_formatter, name, additional_flags), + } + ) + return resolved_dependencies + + @classmethod + def build( + cls, + model: SimpleRetrieverModel, + config: Config, + dependency_constructor: Callable[..., Any], + additional_flags: AdditionalFlags, + **kwargs: Any, + ) -> "SimpleRetriever": + """ + Builds up the Component and it's component-specific dependencies. + Order of operations: + - build the dependencies first + - build the component with the resolved dependencies + """ + resolved_dependencies: Mapping[str, Any] = cls.resolve_dependencies( + model=model, + config=config, + dependency_constructor=dependency_constructor, + additional_flags=additional_flags, + **kwargs, + ) + if cls._should_use_lazy_simple_retriever: + LazySimpleRetriever(**resolved_dependencies) + return SimpleRetriever(**resolved_dependencies) def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._paginator = self.paginator or NoPagination(parameters=parameters) From 3cdd4260123c19e604955fcf69d933dd0ad75923 Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Fri, 25 Jul 2025 13:24:22 +0000 Subject: [PATCH 12/21] Auto-fix lint and format issues --- .../declarative/extractors/record_filter.py | 2 +- .../parsers/component_constructor.py | 5 +- .../parsers/model_to_component_factory.py | 2 +- .../query_properties/query_properties.py | 14 ++-- .../retrievers/simple_retriever.py | 76 +++++++++---------- 5 files changed, 49 insertions(+), 50 deletions(-) diff --git a/airbyte_cdk/sources/declarative/extractors/record_filter.py b/airbyte_cdk/sources/declarative/extractors/record_filter.py index 809a190bcd..3816e57ab3 100644 --- a/airbyte_cdk/sources/declarative/extractors/record_filter.py +++ b/airbyte_cdk/sources/declarative/extractors/record_filter.py @@ -14,8 +14,8 @@ RecordFilter as RecordFilterModel, ) from airbyte_cdk.sources.declarative.parsers.component_constructor import ( - ComponentConstructor, AdditionalFlags, + ComponentConstructor, ) from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState diff --git a/airbyte_cdk/sources/declarative/parsers/component_constructor.py b/airbyte_cdk/sources/declarative/parsers/component_constructor.py index 7ee9c3546b..7ffdac822c 100644 --- a/airbyte_cdk/sources/declarative/parsers/component_constructor.py +++ b/airbyte_cdk/sources/declarative/parsers/component_constructor.py @@ -7,12 +7,11 @@ from pydantic.v1 import BaseModel -from airbyte_cdk.sources.message import MessageRepository +from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType +from airbyte_cdk.sources.message import MessageRepository from airbyte_cdk.sources.types import Config -from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager - M = TypeVar("M", bound=BaseModel) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 02f262ae5c..102d6829ca 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -448,8 +448,8 @@ ZipfileDecoder as ZipfileDecoderModel, ) from airbyte_cdk.sources.declarative.parsers.component_constructor import ( - ComponentConstructor, AdditionalFlags, + ComponentConstructor, ) from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import ( COMPONENTS_MODULE_NAME, diff --git a/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py b/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py index e91f831e0c..051cb4943c 100644 --- a/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py +++ b/airbyte_cdk/sources/declarative/requesters/query_properties/query_properties.py @@ -1,20 +1,20 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. from dataclasses import InitVar, dataclass -from typing import Any, Iterable, List, Mapping, Optional, Union, Callable +from typing import Any, Callable, Iterable, List, Mapping, Optional, Union -from airbyte_cdk.sources.declarative.requesters.query_properties import ( - PropertiesFromEndpoint, - PropertyChunking, -) -from airbyte_cdk.sources.types import Config, StreamSlice from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( QueryProperties as QueryPropertiesModel, ) from airbyte_cdk.sources.declarative.parsers.component_constructor import ( - ComponentConstructor, AdditionalFlags, + ComponentConstructor, +) +from airbyte_cdk.sources.declarative.requesters.query_properties import ( + PropertiesFromEndpoint, + PropertyChunking, ) +from airbyte_cdk.sources.types import Config, StreamSlice @dataclass diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index cf8cb0a3ee..95eb37b83a 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -10,6 +10,7 @@ from typing import ( Any, Callable, + Dict, Iterable, List, Mapping, @@ -18,73 +19,72 @@ Set, Tuple, Union, - Dict, cast, ) import requests +from pydantic import BaseModel +from requests import Response from typing_extensions import deprecated from airbyte_cdk.models import AirbyteMessage +from airbyte_cdk.sources.declarative.decoders import JsonDecoder from airbyte_cdk.sources.declarative.extractors.http_selector import HttpSelector -from airbyte_cdk.sources.declarative.incremental import ResumableFullRefreshCursor +from airbyte_cdk.sources.declarative.incremental import ( + DatetimeBasedCursor, + ResumableFullRefreshCursor, +) from airbyte_cdk.sources.declarative.incremental.declarative_cursor import DeclarativeCursor from airbyte_cdk.sources.declarative.interpolation import InterpolatedString -from airbyte_cdk.sources.declarative.partition_routers.single_partition_router import ( - SinglePartitionRouter, -) -from airbyte_cdk.sources.declarative.requesters.paginators.no_pagination import NoPagination -from airbyte_cdk.sources.declarative.requesters.paginators.paginator import Paginator -from airbyte_cdk.sources.declarative.requesters.query_properties import QueryProperties -from airbyte_cdk.sources.declarative.requesters.request_options import ( - DefaultRequestOptionsProvider, - RequestOptionsProvider, -) -from airbyte_cdk.sources.declarative.requesters.requester import Requester -from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever -from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer -from airbyte_cdk.sources.http_logger import format_http_message -from airbyte_cdk.sources.source import ExperimentalClassWarning -from airbyte_cdk.sources.streams.core import StreamData -from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState -from airbyte_cdk.utils.mapping_helpers import combine_mappings from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( - SimpleRetriever as SimpleRetrieverModel, -) -from airbyte_cdk.sources.declarative.parsers.component_constructor import ( - ComponentConstructor, - AdditionalFlags, + CustomIncrementalSync as CustomIncrementalSyncModel, ) -from airbyte_cdk.sources.declarative.transformations import RecordTransformation -from airbyte_cdk.sources.declarative.decoders import JsonDecoder -from pydantic import BaseModel -from airbyte_cdk.sources.declarative.retrievers.file_uploader import DefaultFileUploader from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( - IncrementingCountCursor as IncrementingCountCursorModel, + CustomRequester as CustomRequesterModel, ) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( DatetimeBasedCursor as DatetimeBasedCursorModel, ) - -from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( - CustomIncrementalSync as CustomIncrementalSyncModel, -) -from requests import Response from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( HttpRequester as HttpRequesterModel, ) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( - CustomRequester as CustomRequesterModel, + IncrementingCountCursor as IncrementingCountCursorModel, ) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( QueryProperties as QueryPropertiesModel, ) -from airbyte_cdk.sources.declarative.requesters.query_properties import QueryProperties -from airbyte_cdk.sources.declarative.incremental import DatetimeBasedCursor +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + SimpleRetriever as SimpleRetrieverModel, +) from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( SubstreamPartitionRouter as SubstreamPartitionRouterModel, ) +from airbyte_cdk.sources.declarative.parsers.component_constructor import ( + AdditionalFlags, + ComponentConstructor, +) +from airbyte_cdk.sources.declarative.partition_routers.single_partition_router import ( + SinglePartitionRouter, +) +from airbyte_cdk.sources.declarative.requesters.paginators.no_pagination import NoPagination +from airbyte_cdk.sources.declarative.requesters.paginators.paginator import Paginator +from airbyte_cdk.sources.declarative.requesters.query_properties import QueryProperties +from airbyte_cdk.sources.declarative.requesters.request_options import ( + DefaultRequestOptionsProvider, + RequestOptionsProvider, +) +from airbyte_cdk.sources.declarative.requesters.requester import Requester +from airbyte_cdk.sources.declarative.retrievers.file_uploader import DefaultFileUploader +from airbyte_cdk.sources.declarative.retrievers.retriever import Retriever from airbyte_cdk.sources.declarative.stream_slicers import StreamSlicerTestReadDecorator +from airbyte_cdk.sources.declarative.stream_slicers.stream_slicer import StreamSlicer +from airbyte_cdk.sources.declarative.transformations import RecordTransformation +from airbyte_cdk.sources.http_logger import format_http_message +from airbyte_cdk.sources.source import ExperimentalClassWarning +from airbyte_cdk.sources.streams.core import StreamData +from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState +from airbyte_cdk.utils.mapping_helpers import combine_mappings FULL_REFRESH_SYNC_COMPLETE_KEY = "__ab_full_refresh_sync_complete" From 06e05b9abc0a4b9b3d48108fe7115f919e9fe35c Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 16:55:21 +0300 Subject: [PATCH 13/21] updated check for name --- airbyte_cdk/sources/declarative/retrievers/simple_retriever.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 95eb37b83a..4be780ce64 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -200,7 +200,7 @@ def resolve_dependencies( log_formatter: Optional[Callable[[Response], Any]] = None, **kwargs: Any, ) -> Mapping[str, Any]: - if not name: + if name is None: raise ValueError(f"name argument is required to instance a {cls.__name__}") def _get_url() -> str: From 7d067199e88ecacd9d887f53759f0f3c15ac5839 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 17:23:13 +0300 Subject: [PATCH 14/21] fix unit tests --- .../parsers/model_to_component_factory.py | 18 ++++++++++++++++++ .../retrievers/simple_retriever.py | 19 +++++++++---------- 2 files changed, 27 insertions(+), 10 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 102d6829ca..be99faf1c4 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3146,6 +3146,24 @@ def create_legacy_session_token_authenticator( parameters=model.parameters or {}, ) + def _get_log_formatter( + self, log_formatter: Callable[[Response], Any] | None, name: str + ) -> Callable[[Response], Any] | None: + if self._should_limit_slices_fetched(): + return ( + ( + lambda response: format_http_message( + response, + f"Stream '{name}' request", + f"Request performed in order to extract records for stream '{name}'", + name, + ) + ) + if not log_formatter + else log_formatter + ) + return None + def _should_limit_slices_fetched(self) -> bool: """ Returns True if the number of slices fetched should be limited, False otherwise. diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 4be780ce64..4cb2ad7445 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -132,10 +132,9 @@ class SimpleRetriever(Retriever, ComponentConstructor[SimpleRetrieverModel]): ignore_stream_slicer_parameters_on_paginated_requests: bool = False additional_query_properties: Optional[QueryProperties] = None log_formatter: Optional[Callable[[requests.Response], Any]] = None - _should_use_lazy_simple_retriever: bool = False @classmethod - def _validate_if_lazy_simple_retriever_should_be_applied( + def _should_use_lazy_simple_retriever( cls, name: str, model: SimpleRetrieverModel, @@ -170,7 +169,6 @@ def _validate_if_lazy_simple_retriever_should_be_applied( raise ValueError( f"LazySimpleRetriever only supports JsonDecoder. Found: {model.decoder.type}." ) - cls._should_use_lazy_simple_retriever = True return True return False @@ -312,8 +310,6 @@ def _get_log_formatter( query_properties = dependency_constructor( model=query_properties_definition, config=config, - dependency_constructor=dependency_constructor, - additional_flags=additional_flags, **kwargs, ) @@ -321,8 +317,6 @@ def _get_log_formatter( query_properties = dependency_constructor( model=model.requester.query_properties, config=config, - dependency_constructor=dependency_constructor, - additional_flags=additional_flags, **kwargs, ) @@ -392,7 +386,7 @@ def _get_log_formatter( "parameters": model.parameters or {}, } - if cls._validate_if_lazy_simple_retriever_should_be_applied( + if cls._should_use_lazy_simple_retriever( name, model, additional_flags, incremental_sync ): return resolved_dependencies @@ -427,8 +421,13 @@ def build( additional_flags=additional_flags, **kwargs, ) - if cls._should_use_lazy_simple_retriever: - LazySimpleRetriever(**resolved_dependencies) + if cls._should_use_lazy_simple_retriever( + name=kwargs.get("name"), + model=model, + additional_flags=additional_flags, + incremental_sync=kwargs.get("incremental_sync"), + ): + return LazySimpleRetriever(**resolved_dependencies) return SimpleRetriever(**resolved_dependencies) def __post_init__(self, parameters: Mapping[str, Any]) -> None: From 218dabb57c100d9bcd5a84da73d89f6d37fec54e Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Fri, 25 Jul 2025 14:26:37 +0000 Subject: [PATCH 15/21] Auto-fix lint and format issues --- .../parsers/model_to_component_factory.py | 2 +- .../declarative/retrievers/simple_retriever.py | 12 +++++------- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index be99faf1c4..6593f7dbaa 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3147,7 +3147,7 @@ def create_legacy_session_token_authenticator( ) def _get_log_formatter( - self, log_formatter: Callable[[Response], Any] | None, name: str + self, log_formatter: Callable[[Response], Any] | None, name: str ) -> Callable[[Response], Any] | None: if self._should_limit_slices_fetched(): return ( diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 4cb2ad7445..ee7c8ced76 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -386,9 +386,7 @@ def _get_log_formatter( "parameters": model.parameters or {}, } - if cls._should_use_lazy_simple_retriever( - name, model, additional_flags, incremental_sync - ): + if cls._should_use_lazy_simple_retriever(name, model, additional_flags, incremental_sync): return resolved_dependencies resolved_dependencies.update( @@ -422,10 +420,10 @@ def build( **kwargs, ) if cls._should_use_lazy_simple_retriever( - name=kwargs.get("name"), - model=model, - additional_flags=additional_flags, - incremental_sync=kwargs.get("incremental_sync"), + name=kwargs.get("name"), + model=model, + additional_flags=additional_flags, + incremental_sync=kwargs.get("incremental_sync"), ): return LazySimpleRetriever(**resolved_dependencies) return SimpleRetriever(**resolved_dependencies) From b212ed93694f1e1cfddd719e028840925c7018b0 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Fri, 25 Jul 2025 17:35:17 +0300 Subject: [PATCH 16/21] fix mypy --- .../sources/declarative/retrievers/simple_retriever.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index ee7c8ced76..1259c2463e 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -136,7 +136,6 @@ class SimpleRetriever(Retriever, ComponentConstructor[SimpleRetrieverModel]): @classmethod def _should_use_lazy_simple_retriever( cls, - name: str, model: SimpleRetrieverModel, additional_flags: AdditionalFlags, incremental_sync: Optional[ @@ -144,7 +143,10 @@ def _should_use_lazy_simple_retriever( IncrementingCountCursorModel, DatetimeBasedCursorModel, CustomIncrementalSyncModel ] ] = None, + name: Optional[str] = None, ) -> bool: + if name is None: + raise ValueError(f"name argument is required to instance a {cls.__name__}") if ( model.partition_router and isinstance(model.partition_router, SubstreamPartitionRouterModel) @@ -386,7 +388,7 @@ def _get_log_formatter( "parameters": model.parameters or {}, } - if cls._should_use_lazy_simple_retriever(name, model, additional_flags, incremental_sync): + if cls._should_use_lazy_simple_retriever(model, additional_flags, incremental_sync, name): return resolved_dependencies resolved_dependencies.update( @@ -420,10 +422,10 @@ def build( **kwargs, ) if cls._should_use_lazy_simple_retriever( - name=kwargs.get("name"), model=model, additional_flags=additional_flags, incremental_sync=kwargs.get("incremental_sync"), + name=kwargs.get("name"), ): return LazySimpleRetriever(**resolved_dependencies) return SimpleRetriever(**resolved_dependencies) From b9e26bd1515c3d8d7d2d73787089bc80e6bd0dd9 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Tue, 5 Aug 2025 16:35:58 +0300 Subject: [PATCH 17/21] updated simple retriever --- .../sources/declarative/retrievers/simple_retriever.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index f15fd28608..79fa0858dd 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -85,6 +85,8 @@ from airbyte_cdk.sources.streams.core import StreamData from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState from airbyte_cdk.utils.mapping_helpers import combine_mappings +from airbyte_cdk.sources.streams.concurrent.cursor import Cursor + FULL_REFRESH_SYNC_COMPLETE_KEY = "__ab_full_refresh_sync_complete" @@ -187,7 +189,7 @@ def resolve_dependencies( primary_key: Optional[Union[str, List[str], List[List[str]]]] = None, stream_slicer: Optional[StreamSlicer] = None, request_options_provider: Optional[RequestOptionsProvider] = None, - stop_condition_on_cursor: bool = False, + stop_condition_cursor: Optional[Cursor] = None, client_side_incremental_sync: Optional[Dict[str, Any]] = None, transformations: Optional[List[RecordTransformation]] = None, file_uploader: Optional[DefaultFileUploader] = None, @@ -356,7 +358,6 @@ def _get_log_formatter( ), ) - cursor_used_for_stop_condition = cursor if stop_condition_on_cursor else None paginator = ( dependency_constructor( model=model.paginator, @@ -364,7 +365,7 @@ def _get_log_formatter( url_base=_get_url(), extractor_model=model.record_selector.extractor, decoder=decoder, - cursor_used_for_stop_condition=cursor_used_for_stop_condition, + cursor_used_for_stop_condition=stop_condition_cursor or None, ) if model.paginator else NoPagination(parameters={}) From 1b02cdca0f7ac56e0dfef6ce93c99345f1b4d98a Mon Sep 17 00:00:00 2001 From: octavia-squidington-iii Date: Tue, 5 Aug 2025 13:43:23 +0000 Subject: [PATCH 18/21] Auto-fix lint and format issues --- airbyte_cdk/sources/declarative/retrievers/simple_retriever.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 79fa0858dd..371b4edb2e 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -82,11 +82,10 @@ from airbyte_cdk.sources.declarative.transformations import RecordTransformation from airbyte_cdk.sources.http_logger import format_http_message from airbyte_cdk.sources.source import ExperimentalClassWarning +from airbyte_cdk.sources.streams.concurrent.cursor import Cursor from airbyte_cdk.sources.streams.core import StreamData from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState from airbyte_cdk.utils.mapping_helpers import combine_mappings -from airbyte_cdk.sources.streams.concurrent.cursor import Cursor - FULL_REFRESH_SYNC_COMPLETE_KEY = "__ab_full_refresh_sync_complete" From 5e71790a32ef4f9654edab05ee28da342f705392 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Tue, 12 Aug 2025 13:30:30 +0300 Subject: [PATCH 19/21] deleted _json_schema_type_name_to_type from ComponentConstructor class --- .../declarative/parsers/component_constructor.py | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/airbyte_cdk/sources/declarative/parsers/component_constructor.py b/airbyte_cdk/sources/declarative/parsers/component_constructor.py index 7ffdac822c..11cad49ba3 100644 --- a/airbyte_cdk/sources/declarative/parsers/component_constructor.py +++ b/airbyte_cdk/sources/declarative/parsers/component_constructor.py @@ -88,15 +88,3 @@ def build( # returns the instance of the component class, # with resolved dependencies and model-specific arguments. return cls(**resolved_dependencies) - - @staticmethod - def _json_schema_type_name_to_type(value_type: Optional[ValueType]) -> Optional[Type[Any]]: - if not value_type: - return None - names_to_types = { - ValueType.string: str, - ValueType.number: float, - ValueType.integer: int, - ValueType.boolean: bool, - } - return names_to_types[value_type] From 906d5878fb8ae31517471f92d569bef80d665b21 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Thu, 11 Sep 2025 14:36:02 +0300 Subject: [PATCH 20/21] format fix --- .../declarative/retrievers/simple_retriever.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py index 6dde13ce47..37592b0e43 100644 --- a/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py +++ b/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py @@ -140,13 +140,13 @@ def _should_use_lazy_simple_retriever( name: Optional[str] = None, ) -> bool: if ( - model.partition_router - and isinstance(model.partition_router, SubstreamPartitionRouterModel) - and not bool(additional_flags.connector_state_manager.get_stream_state(name, None)) - and any( - parent_stream_config.lazy_read_pointer - for parent_stream_config in model.partition_router.parent_stream_configs - ) + model.partition_router + and isinstance(model.partition_router, SubstreamPartitionRouterModel) + and not bool(additional_flags.connector_state_manager.get_stream_state(name, None)) + and any( + parent_stream_config.lazy_read_pointer + for parent_stream_config in model.partition_router.parent_stream_configs + ) ): if incremental_sync: if incremental_sync.type != "DatetimeBasedCursor": @@ -324,7 +324,7 @@ def _get_log_formatter( if not request_options_provider: request_options_provider = DefaultRequestOptionsProvider(parameters={}) if isinstance(request_options_provider, DefaultRequestOptionsProvider) and isinstance( - partition_router, PartitionRouter + partition_router, PartitionRouter ): request_options_provider = partition_router @@ -342,7 +342,7 @@ def _get_log_formatter( ) ignore_stream_slicer_parameters_on_paginated_requests = ( - model.ignore_stream_slicer_parameters_on_paginated_requests or False + model.ignore_stream_slicer_parameters_on_paginated_requests or False ) resolved_dependencies = { From 2fca25a6bf5f905023dbdf613551199804014a83 Mon Sep 17 00:00:00 2001 From: darynaishchenko Date: Thu, 11 Sep 2025 15:58:04 +0300 Subject: [PATCH 21/21] fix imports --- .../sources/declarative/parsers/model_to_component_factory.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 7692f92327..b68cddc942 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -632,6 +632,10 @@ from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction from airbyte_cdk.sources.types import Config from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer +from airbyte_cdk.sources.declarative.parsers.component_constructor import ( + AdditionalFlags, + ComponentConstructor, +) ComponentDefinition = Mapping[str, Any]