airbyte_cdk.sources.declarative.parsers.model_to_component_factory

   1#
   2# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
   3#
   4
   5from __future__ import annotations
   6
   7import datetime
   8import importlib
   9import inspect
  10import json
  11import logging
  12import re
  13from functools import partial
  14from typing import (
  15    TYPE_CHECKING,
  16    Any,
  17    Callable,
  18    Dict,
  19    List,
  20    Mapping,
  21    MutableMapping,
  22    Optional,
  23    Tuple,
  24    Type,
  25    Union,
  26    cast,
  27    get_args,
  28    get_origin,
  29    get_type_hints,
  30)
  31
  32if TYPE_CHECKING:
  33    from airbyte_cdk.legacy.sources.declarative.incremental.datetime_based_cursor import (
  34        DatetimeBasedCursor,
  35    )
  36
  37from airbyte_protocol_dataclasses.models import ConfiguredAirbyteStream
  38from isodate import parse_duration
  39from pydantic.v1 import BaseModel
  40from requests import Response
  41
  42from airbyte_cdk.connector_builder.models import (
  43    LogMessage as ConnectorBuilderLogMessage,
  44)
  45from airbyte_cdk.models import (
  46    AirbyteStateBlob,
  47    AirbyteStateMessage,
  48    AirbyteStateType,
  49    AirbyteStreamState,
  50    ConfiguredAirbyteCatalog,
  51    FailureType,
  52    Level,
  53    StreamDescriptor,
  54)
  55from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager
  56from airbyte_cdk.sources.declarative.async_job.job_orchestrator import AsyncJobOrchestrator
  57from airbyte_cdk.sources.declarative.async_job.job_tracker import JobTracker
  58from airbyte_cdk.sources.declarative.async_job.repository import AsyncJobRepository
  59from airbyte_cdk.sources.declarative.async_job.status import AsyncJobStatus
  60from airbyte_cdk.sources.declarative.auth import DeclarativeOauth2Authenticator, JwtAuthenticator
  61from airbyte_cdk.sources.declarative.auth.declarative_authenticator import (
  62    DeclarativeAuthenticator,
  63    NoAuth,
  64)
  65from airbyte_cdk.sources.declarative.auth.jwt import JwtAlgorithm
  66from airbyte_cdk.sources.declarative.auth.oauth import (
  67    DeclarativeSingleUseRefreshTokenOauth2Authenticator,
  68)
  69from airbyte_cdk.sources.declarative.auth.rate_limited_multiple_token import (
  70    RateLimitedMultipleTokenAuthenticator,
  71    TokenQuota,
  72)
  73from airbyte_cdk.sources.declarative.auth.selective_authenticator import SelectiveAuthenticator
  74from airbyte_cdk.sources.declarative.auth.token import (
  75    ApiKeyAuthenticator,
  76    BasicHttpAuthenticator,
  77    BearerAuthenticator,
  78    LegacySessionTokenAuthenticator,
  79)
  80from airbyte_cdk.sources.declarative.auth.token_provider import (
  81    InterpolatedSessionTokenProvider,
  82    InterpolatedStringTokenProvider,
  83    SessionTokenProvider,
  84    TokenProvider,
  85)
  86from airbyte_cdk.sources.declarative.checks import (
  87    CheckDynamicStream,
  88    CheckStream,
  89    DynamicStreamCheckConfig,
  90)
  91from airbyte_cdk.sources.declarative.concurrency_level import ConcurrencyLevel
  92from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime
  93from airbyte_cdk.sources.declarative.decoders import (
  94    Decoder,
  95    IterableDecoder,
  96    JsonDecoder,
  97    PaginationDecoderDecorator,
  98    XmlDecoder,
  99    ZipfileDecoder,
 100)
 101from airbyte_cdk.sources.declarative.decoders.composite_raw_decoder import (
 102    CompositeRawDecoder,
 103    CsvParser,
 104    GzipParser,
 105    JsonItemsParser,
 106    JsonLineParser,
 107    JsonParser,
 108    Parser,
 109)
 110from airbyte_cdk.sources.declarative.expanders.record_expander import (
 111    OnNoRecords,
 112    RecordExpander,
 113)
 114from airbyte_cdk.sources.declarative.extractors import (
 115    DpathExtractor,
 116    RecordFilter,
 117    RecordSelector,
 118    ResponseToFileExtractor,
 119)
 120from airbyte_cdk.sources.declarative.extractors.record_extractor import RecordExtractor
 121from airbyte_cdk.sources.declarative.extractors.record_filter import (
 122    ClientSideIncrementalRecordFilterDecorator,
 123)
 124from airbyte_cdk.sources.declarative.incremental import (
 125    ConcurrentCursorFactory,
 126    ConcurrentPerPartitionCursor,
 127)
 128from airbyte_cdk.sources.declarative.interpolation import InterpolatedString
 129from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
 130from airbyte_cdk.sources.declarative.migrations.legacy_to_per_partition_state_migration import (
 131    LegacyToPerPartitionStateMigration,
 132)
 133from airbyte_cdk.sources.declarative.models import (
 134    CustomStateMigration,
 135    PaginationResetLimits,
 136)
 137from airbyte_cdk.sources.declarative.models.base_model_with_deprecations import (
 138    DEPRECATION_LOGS_TAG,
 139    BaseModelWithDeprecations,
 140)
 141from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 142    Action1 as PaginationResetActionModel,
 143)
 144from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 145    AddedFieldDefinition as AddedFieldDefinitionModel,
 146)
 147from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 148    AddFields as AddFieldsModel,
 149)
 150from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 151    ApiKeyAuthenticator as ApiKeyAuthenticatorModel,
 152)
 153from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 154    AsyncJobStatusMap as AsyncJobStatusMapModel,
 155)
 156from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 157    AsyncRetriever as AsyncRetrieverModel,
 158)
 159from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 160    BasicHttpAuthenticator as BasicHttpAuthenticatorModel,
 161)
 162from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 163    BearerAuthenticator as BearerAuthenticatorModel,
 164)
 165from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 166    CheckDynamicStream as CheckDynamicStreamModel,
 167)
 168from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 169    CheckStream as CheckStreamModel,
 170)
 171from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 172    ComplexFieldType as ComplexFieldTypeModel,
 173)
 174from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 175    ComponentMappingDefinition as ComponentMappingDefinitionModel,
 176)
 177from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 178    CompositeErrorHandler as CompositeErrorHandlerModel,
 179)
 180from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 181    ConcurrencyLevel as ConcurrencyLevelModel,
 182)
 183from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 184    ConfigAddFields as ConfigAddFieldsModel,
 185)
 186from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 187    ConfigComponentsResolver as ConfigComponentsResolverModel,
 188)
 189from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 190    ConfigMigration as ConfigMigrationModel,
 191)
 192from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 193    ConfigRemapField as ConfigRemapFieldModel,
 194)
 195from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 196    ConfigRemoveFields as ConfigRemoveFieldsModel,
 197)
 198from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 199    ConstantBackoffStrategy as ConstantBackoffStrategyModel,
 200)
 201from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 202    CsvDecoder as CsvDecoderModel,
 203)
 204from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 205    CursorPagination as CursorPaginationModel,
 206)
 207from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 208    CustomAuthenticator as CustomAuthenticatorModel,
 209)
 210from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 211    CustomBackoffStrategy as CustomBackoffStrategyModel,
 212)
 213from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 214    CustomConfigTransformation as CustomConfigTransformationModel,
 215)
 216from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 217    CustomDecoder as CustomDecoderModel,
 218)
 219from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 220    CustomErrorHandler as CustomErrorHandlerModel,
 221)
 222from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 223    CustomPaginationStrategy as CustomPaginationStrategyModel,
 224)
 225from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 226    CustomPartitionRouter as CustomPartitionRouterModel,
 227)
 228from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 229    CustomRecordExtractor as CustomRecordExtractorModel,
 230)
 231from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 232    CustomRecordFilter as CustomRecordFilterModel,
 233)
 234from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 235    CustomRequester as CustomRequesterModel,
 236)
 237from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 238    CustomRetriever as CustomRetrieverModel,
 239)
 240from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 241    CustomSchemaLoader as CustomSchemaLoader,
 242)
 243from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 244    CustomSchemaNormalization as CustomSchemaNormalizationModel,
 245)
 246from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 247    CustomTransformation as CustomTransformationModel,
 248)
 249from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 250    CustomValidationStrategy as CustomValidationStrategyModel,
 251)
 252from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 253    DatetimeBasedCursor as DatetimeBasedCursorModel,
 254)
 255from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 256    DeclarativeStream as DeclarativeStreamModel,
 257)
 258from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 259    DefaultErrorHandler as DefaultErrorHandlerModel,
 260)
 261from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 262    DefaultPaginator as DefaultPaginatorModel,
 263)
 264from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 265    DpathExtractor as DpathExtractorModel,
 266)
 267from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 268    DpathFlattenFields as DpathFlattenFieldsModel,
 269)
 270from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 271    DpathValidator as DpathValidatorModel,
 272)
 273from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 274    DynamicSchemaLoader as DynamicSchemaLoaderModel,
 275)
 276from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 277    DynamicStreamCheckConfig as DynamicStreamCheckConfigModel,
 278)
 279from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 280    ExponentialBackoffStrategy as ExponentialBackoffStrategyModel,
 281)
 282from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 283    FileUploader as FileUploaderModel,
 284)
 285from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 286    FixedWindowCallRatePolicy as FixedWindowCallRatePolicyModel,
 287)
 288from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 289    FlattenFields as FlattenFieldsModel,
 290)
 291from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 292    GroupByKeyMergeStrategy as GroupByKeyMergeStrategyModel,
 293)
 294from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 295    GroupingPartitionRouter as GroupingPartitionRouterModel,
 296)
 297from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 298    GzipDecoder as GzipDecoderModel,
 299)
 300from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 301    HTTPAPIBudget as HTTPAPIBudgetModel,
 302)
 303from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 304    HttpComponentsResolver as HttpComponentsResolverModel,
 305)
 306from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 307    HttpRequester as HttpRequesterModel,
 308)
 309from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 310    HttpRequestRegexMatcher as HttpRequestRegexMatcherModel,
 311)
 312from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 313    HttpResponseFilter as HttpResponseFilterModel,
 314)
 315from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 316    IncrementingCountCursor as IncrementingCountCursorModel,
 317)
 318from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 319    InlineSchemaLoader as InlineSchemaLoaderModel,
 320)
 321from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 322    IterableDecoder as IterableDecoderModel,
 323)
 324from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 325    JsonDecoder as JsonDecoderModel,
 326)
 327from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 328    JsonFileSchemaLoader as JsonFileSchemaLoaderModel,
 329)
 330from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 331    JsonItemsDecoder as JsonItemsDecoderModel,
 332)
 333from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 334    JsonlDecoder as JsonlDecoderModel,
 335)
 336from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 337    JsonSchemaPropertySelector as JsonSchemaPropertySelectorModel,
 338)
 339from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 340    JwtAuthenticator as JwtAuthenticatorModel,
 341)
 342from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 343    JwtHeaders as JwtHeadersModel,
 344)
 345from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 346    JwtPayload as JwtPayloadModel,
 347)
 348from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 349    KeysReplace as KeysReplaceModel,
 350)
 351from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 352    KeysToLower as KeysToLowerModel,
 353)
 354from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 355    KeysToSnakeCase as KeysToSnakeCaseModel,
 356)
 357from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 358    LegacySessionTokenAuthenticator as LegacySessionTokenAuthenticatorModel,
 359)
 360from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 361    LegacyToPerPartitionStateMigration as LegacyToPerPartitionStateMigrationModel,
 362)
 363from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 364    ListPartitionRouter as ListPartitionRouterModel,
 365)
 366from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 367    MinMaxDatetime as MinMaxDatetimeModel,
 368)
 369from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 370    MovingWindowCallRatePolicy as MovingWindowCallRatePolicyModel,
 371)
 372from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 373    NoAuth as NoAuthModel,
 374)
 375from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 376    NoPagination as NoPaginationModel,
 377)
 378from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 379    OAuthAuthenticator as OAuthAuthenticatorModel,
 380)
 381from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 382    OffsetIncrement as OffsetIncrementModel,
 383)
 384from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 385    PageIncrement as PageIncrementModel,
 386)
 387from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 388    PaginationReset as PaginationResetModel,
 389)
 390from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 391    ParametrizedComponentsResolver as ParametrizedComponentsResolverModel,
 392)
 393from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 394    ParentStreamConfig as ParentStreamConfigModel,
 395)
 396from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 397    PredicateValidator as PredicateValidatorModel,
 398)
 399from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 400    PropertiesFromEndpoint as PropertiesFromEndpointModel,
 401)
 402from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 403    PropertyChunking as PropertyChunkingModel,
 404)
 405from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 406    PropertyLimitType as PropertyLimitTypeModel,
 407)
 408from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 409    QueryProperties as QueryPropertiesModel,
 410)
 411from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 412    Rate as RateModel,
 413)
 414from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 415    RateLimitedMultipleTokenAuthenticator as RateLimitedMultipleTokenAuthenticatorModel,
 416)
 417from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 418    RecordExpander as RecordExpanderModel,
 419)
 420from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 421    RecordFilter as RecordFilterModel,
 422)
 423from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 424    RecordSelector as RecordSelectorModel,
 425)
 426from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 427    RefreshTokenUpdater as RefreshTokenUpdaterModel,
 428)
 429from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 430    RemoveFields as RemoveFieldsModel,
 431)
 432from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 433    RequestOption as RequestOptionModel,
 434)
 435from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 436    RequestPath as RequestPathModel,
 437)
 438from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 439    ResponseToFileExtractor as ResponseToFileExtractorModel,
 440)
 441from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 442    SchemaNormalization as SchemaNormalizationModel,
 443)
 444from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 445    SchemaTypeIdentifier as SchemaTypeIdentifierModel,
 446)
 447from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 448    SelectiveAuthenticator as SelectiveAuthenticatorModel,
 449)
 450from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 451    SessionTokenAuthenticator as SessionTokenAuthenticatorModel,
 452)
 453from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 454    SimpleRetriever as SimpleRetrieverModel,
 455)
 456from airbyte_cdk.sources.declarative.models.declarative_component_schema import Spec as SpecModel
 457from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 458    StateDelegatingStream as StateDelegatingStreamModel,
 459)
 460from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 461    StreamConfig as StreamConfigModel,
 462)
 463from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 464    SubstreamPartitionRouter as SubstreamPartitionRouterModel,
 465)
 466from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 467    TypesMap as TypesMapModel,
 468)
 469from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 470    UnionPartitionRouter as UnionPartitionRouterModel,
 471)
 472from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 473    UnlimitedCallRatePolicy as UnlimitedCallRatePolicyModel,
 474)
 475from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 476    ValidateAdheresToSchema as ValidateAdheresToSchemaModel,
 477)
 478from airbyte_cdk.sources.declarative.models.declarative_component_schema import ValueType
 479from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 480    WaitTimeFromHeader as WaitTimeFromHeaderModel,
 481)
 482from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 483    WaitUntilTimeFromHeader as WaitUntilTimeFromHeaderModel,
 484)
 485from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 486    XmlDecoder as XmlDecoderModel,
 487)
 488from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 489    ZipfileDecoder as ZipfileDecoderModel,
 490)
 491from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
 492    INJECTED_MANIFEST,
 493    AirbyteCustomCodeNotPermittedError,
 494    custom_code_execution_permitted,
 495)
 496from airbyte_cdk.sources.declarative.partition_routers import (
 497    CartesianProductStreamSlicer,
 498    GroupingPartitionRouter,
 499    ListPartitionRouter,
 500    PartitionRouter,
 501    SinglePartitionRouter,
 502    SubstreamPartitionRouter,
 503    UnionPartitionRouter,
 504)
 505from airbyte_cdk.sources.declarative.partition_routers.async_job_partition_router import (
 506    AsyncJobPartitionRouter,
 507)
 508from airbyte_cdk.sources.declarative.partition_routers.substream_partition_router import (
 509    ParentStreamConfig,
 510)
 511from airbyte_cdk.sources.declarative.requesters import HttpRequester, RequestOption
 512from airbyte_cdk.sources.declarative.requesters.error_handlers import (
 513    CompositeErrorHandler,
 514    DefaultErrorHandler,
 515    HttpResponseFilter,
 516)
 517from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies import (
 518    ConstantBackoffStrategy,
 519    ExponentialBackoffStrategy,
 520    WaitTimeFromHeaderBackoffStrategy,
 521    WaitUntilTimeFromHeaderBackoffStrategy,
 522)
 523from airbyte_cdk.sources.declarative.requesters.http_job_repository import AsyncHttpJobRepository
 524from airbyte_cdk.sources.declarative.requesters.paginators import (
 525    DefaultPaginator,
 526    NoPagination,
 527    PaginatorTestReadDecorator,
 528)
 529from airbyte_cdk.sources.declarative.requesters.paginators.strategies import (
 530    CursorPaginationStrategy,
 531    CursorStopCondition,
 532    OffsetIncrement,
 533    PageIncrement,
 534    StopConditionPaginationStrategyDecorator,
 535)
 536from airbyte_cdk.sources.declarative.requesters.query_properties import (
 537    PropertiesFromEndpoint,
 538    PropertyChunking,
 539    QueryProperties,
 540)
 541from airbyte_cdk.sources.declarative.requesters.query_properties.property_chunking import (
 542    PropertyLimitType,
 543)
 544from airbyte_cdk.sources.declarative.requesters.query_properties.property_selector import (
 545    JsonSchemaPropertySelector,
 546)
 547from airbyte_cdk.sources.declarative.requesters.query_properties.strategies import (
 548    GroupByKey,
 549)
 550from airbyte_cdk.sources.declarative.requesters.request_option import RequestOptionType
 551from airbyte_cdk.sources.declarative.requesters.request_options import (
 552    DatetimeBasedRequestOptionsProvider,
 553    DefaultRequestOptionsProvider,
 554    InterpolatedRequestOptionsProvider,
 555    RequestOptionsProvider,
 556)
 557from airbyte_cdk.sources.declarative.requesters.request_options.per_partition_request_option_provider import (
 558    PerPartitionRequestOptionsProvider,
 559)
 560from airbyte_cdk.sources.declarative.requesters.request_path import RequestPath
 561from airbyte_cdk.sources.declarative.requesters.requester import HttpMethod, Requester
 562from airbyte_cdk.sources.declarative.resolvers import (
 563    ComponentMappingDefinition,
 564    ConfigComponentsResolver,
 565    HttpComponentsResolver,
 566    ParametrizedComponentsResolver,
 567    StreamConfig,
 568    StreamParametersDefinition,
 569)
 570from airbyte_cdk.sources.declarative.retrievers import (
 571    AsyncRetriever,
 572    LazySimpleRetriever,
 573    SimpleRetriever,
 574)
 575from airbyte_cdk.sources.declarative.retrievers.file_uploader import (
 576    ConnectorBuilderFileUploader,
 577    DefaultFileUploader,
 578    FileUploader,
 579    LocalFileSystemFileWriter,
 580    NoopFileWriter,
 581)
 582from airbyte_cdk.sources.declarative.retrievers.pagination_tracker import PaginationTracker
 583from airbyte_cdk.sources.declarative.schema import (
 584    ComplexFieldType,
 585    DefaultSchemaLoader,
 586    DynamicSchemaLoader,
 587    InlineSchemaLoader,
 588    JsonFileSchemaLoader,
 589    SchemaLoader,
 590    SchemaTypeIdentifier,
 591    TypesMap,
 592)
 593from airbyte_cdk.sources.declarative.schema.caching_schema_loader_decorator import (
 594    CachingSchemaLoaderDecorator,
 595)
 596from airbyte_cdk.sources.declarative.schema.composite_schema_loader import CompositeSchemaLoader
 597from airbyte_cdk.sources.declarative.spec import ConfigMigration, Spec
 598from airbyte_cdk.sources.declarative.stream_slicers import (
 599    StreamSlicer,
 600    StreamSlicerTestReadDecorator,
 601)
 602from airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator import (
 603    DeclarativePartitionFactory,
 604    StreamSlicerPartitionGenerator,
 605)
 606from airbyte_cdk.sources.declarative.transformations import (
 607    AddFields,
 608    RecordTransformation,
 609    RemoveFields,
 610)
 611from airbyte_cdk.sources.declarative.transformations.add_fields import AddedFieldDefinition
 612from airbyte_cdk.sources.declarative.transformations.config_transformations import (
 613    ConfigAddFields,
 614    ConfigRemapField,
 615    ConfigRemoveFields,
 616)
 617from airbyte_cdk.sources.declarative.transformations.config_transformations.config_transformation import (
 618    ConfigTransformation,
 619)
 620from airbyte_cdk.sources.declarative.transformations.dpath_flatten_fields import (
 621    DpathFlattenFields,
 622    KeyTransformation,
 623)
 624from airbyte_cdk.sources.declarative.transformations.flatten_fields import (
 625    FlattenFields,
 626)
 627from airbyte_cdk.sources.declarative.transformations.keys_replace_transformation import (
 628    KeysReplaceTransformation,
 629)
 630from airbyte_cdk.sources.declarative.transformations.keys_to_lower_transformation import (
 631    KeysToLowerTransformation,
 632)
 633from airbyte_cdk.sources.declarative.transformations.keys_to_snake_transformation import (
 634    KeysToSnakeCaseTransformation,
 635)
 636from airbyte_cdk.sources.declarative.validators import (
 637    DpathValidator,
 638    PredicateValidator,
 639    ValidateAdheresToSchema,
 640)
 641from airbyte_cdk.sources.http_logger import format_http_message
 642from airbyte_cdk.sources.message import (
 643    InMemoryMessageRepository,
 644    LogAppenderMessageRepositoryDecorator,
 645    MessageRepository,
 646    NoopMessageRepository,
 647)
 648from airbyte_cdk.sources.message.repository import StateFilteringMessageRepository
 649from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY
 650from airbyte_cdk.sources.streams.call_rate import (
 651    APIBudget,
 652    FixedWindowCallRatePolicy,
 653    HttpAPIBudget,
 654    HttpRequestRegexMatcher,
 655    MovingWindowCallRatePolicy,
 656    Rate,
 657    UnlimitedCallRatePolicy,
 658)
 659from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
 660from airbyte_cdk.sources.streams.concurrent.clamping import (
 661    ClampingEndProvider,
 662    ClampingStrategy,
 663    DayClampingStrategy,
 664    MonthClampingStrategy,
 665    NoClamping,
 666    WeekClampingStrategy,
 667    Weekday,
 668)
 669from airbyte_cdk.sources.streams.concurrent.cursor import (
 670    ConcurrentCursor,
 671    Cursor,
 672    CursorField,
 673    FinalStateCursor,
 674)
 675from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream
 676from airbyte_cdk.sources.streams.concurrent.helpers import get_primary_key_from_stream
 677from airbyte_cdk.sources.streams.concurrent.partitions.stream_slicer import (
 678    StreamSlicer as ConcurrentStreamSlicer,
 679)
 680from airbyte_cdk.sources.streams.concurrent.state_converters.datetime_stream_state_converter import (
 681    CustomFormatConcurrentStreamStateConverter,
 682    DateTimeStreamStateConverter,
 683)
 684from airbyte_cdk.sources.streams.concurrent.state_converters.incrementing_count_stream_state_converter import (
 685    IncrementingCountStreamStateConverter,
 686)
 687from airbyte_cdk.sources.streams.http.error_handlers.response_models import ResponseAction
 688from airbyte_cdk.sources.types import Config
 689from airbyte_cdk.sources.utils.transform import TransformConfig, TypeTransformer
 690
 691ComponentDefinition = Mapping[str, Any]
 692
 693SCHEMA_TRANSFORMER_TYPE_MAPPING = {
 694    SchemaNormalizationModel.None_: TransformConfig.NoTransform,
 695    SchemaNormalizationModel.Default: TransformConfig.DefaultSchemaNormalization,
 696}
 697_NO_STREAM_SLICING = SinglePartitionRouter(parameters={})
 698
 699# Ideally this should use the value defined in ConcurrentDeclarativeSource, but
 700# this would be a circular import
 701MAX_SLICES = 5
 702
 703LOGGER = logging.getLogger(f"airbyte.model_to_component_factory")
 704
 705
 706class ModelToComponentFactory:
 707    EPOCH_DATETIME_FORMAT = "%s"
 708
 709    def __init__(
 710        self,
 711        limit_pages_fetched_per_slice: Optional[int] = None,
 712        limit_slices_fetched: Optional[int] = None,
 713        emit_connector_builder_messages: bool = False,
 714        disable_retries: bool = False,
 715        disable_cache: bool = False,
 716        message_repository: Optional[MessageRepository] = None,
 717        connector_state_manager: Optional[ConnectorStateManager] = None,
 718        max_concurrent_async_job_count: Optional[int] = None,
 719        configured_catalog: Optional[ConfiguredAirbyteCatalog] = None,
 720        api_budget: Optional[APIBudget] = None,
 721        rate_limited_authenticators: Optional[
 722            Dict[str, RateLimitedMultipleTokenAuthenticator]
 723        ] = None,
 724        custom_components_trusted: bool = True,
 725    ):
 726        self._init_mappings()
 727        self._custom_components_trusted = custom_components_trusted
 728        self._limit_pages_fetched_per_slice = limit_pages_fetched_per_slice
 729        self._limit_slices_fetched = limit_slices_fetched
 730        self._emit_connector_builder_messages = emit_connector_builder_messages
 731        self._disable_retries = disable_retries
 732        self._disable_cache = disable_cache
 733        self._message_repository = message_repository or InMemoryMessageRepository(
 734            self._evaluate_log_level(emit_connector_builder_messages)
 735        )
 736        self._stream_name_to_configured_stream = self._create_stream_name_to_configured_stream(
 737            configured_catalog
 738        )
 739        self._connector_state_manager = connector_state_manager or ConnectorStateManager()
 740        self._api_budget: Optional[Union[APIBudget]] = api_budget
 741        # Shared instances so all streams see the same token quota counters (like api_budget)
 742        self._rate_limited_authenticators: Dict[str, RateLimitedMultipleTokenAuthenticator] = (
 743            rate_limited_authenticators if rate_limited_authenticators is not None else {}
 744        )
 745        self._job_tracker: JobTracker = JobTracker(max_concurrent_async_job_count or 1)
 746        # placeholder for deprecation warnings
 747        self._collected_deprecation_logs: List[ConnectorBuilderLogMessage] = []
 748
 749    def _init_mappings(self) -> None:
 750        self.PYDANTIC_MODEL_TO_CONSTRUCTOR: Mapping[Type[BaseModel], Callable[..., Any]] = {
 751            AddedFieldDefinitionModel: self.create_added_field_definition,
 752            AddFieldsModel: self.create_add_fields,
 753            ApiKeyAuthenticatorModel: self.create_api_key_authenticator,
 754            BasicHttpAuthenticatorModel: self.create_basic_http_authenticator,
 755            BearerAuthenticatorModel: self.create_bearer_authenticator,
 756            CheckStreamModel: self.create_check_stream,
 757            DynamicStreamCheckConfigModel: self.create_dynamic_stream_check_config,
 758            CheckDynamicStreamModel: self.create_check_dynamic_stream,
 759            CompositeErrorHandlerModel: self.create_composite_error_handler,
 760            ConcurrencyLevelModel: self.create_concurrency_level,
 761            ConfigMigrationModel: self.create_config_migration,
 762            ConfigAddFieldsModel: self.create_config_add_fields,
 763            ConfigRemapFieldModel: self.create_config_remap_field,
 764            ConfigRemoveFieldsModel: self.create_config_remove_fields,
 765            ConstantBackoffStrategyModel: self.create_constant_backoff_strategy,
 766            CsvDecoderModel: self.create_csv_decoder,
 767            CursorPaginationModel: self.create_cursor_pagination,
 768            CustomAuthenticatorModel: self.create_custom_component,
 769            CustomBackoffStrategyModel: self.create_custom_component,
 770            CustomDecoderModel: self.create_custom_component,
 771            CustomErrorHandlerModel: self.create_custom_component,
 772            CustomRecordExtractorModel: self.create_custom_component,
 773            CustomRecordFilterModel: self.create_custom_component,
 774            CustomRequesterModel: self.create_custom_component,
 775            CustomRetrieverModel: self.create_custom_component,
 776            CustomSchemaLoader: self.create_custom_component,
 777            CustomSchemaNormalizationModel: self.create_custom_component,
 778            CustomStateMigration: self.create_custom_component,
 779            CustomPaginationStrategyModel: self.create_custom_component,
 780            CustomPartitionRouterModel: self.create_custom_component,
 781            CustomTransformationModel: self.create_custom_component,
 782            CustomValidationStrategyModel: self.create_custom_component,
 783            CustomConfigTransformationModel: self.create_custom_component,
 784            DeclarativeStreamModel: self.create_default_stream,
 785            DefaultErrorHandlerModel: self.create_default_error_handler,
 786            DefaultPaginatorModel: self.create_default_paginator,
 787            DpathExtractorModel: self.create_dpath_extractor,
 788            DpathValidatorModel: self.create_dpath_validator,
 789            ResponseToFileExtractorModel: self.create_response_to_file_extractor,
 790            ExponentialBackoffStrategyModel: self.create_exponential_backoff_strategy,
 791            SessionTokenAuthenticatorModel: self.create_session_token_authenticator,
 792            GroupByKeyMergeStrategyModel: self.create_group_by_key,
 793            HttpRequesterModel: self.create_http_requester,
 794            HttpResponseFilterModel: self.create_http_response_filter,
 795            InlineSchemaLoaderModel: self.create_inline_schema_loader,
 796            JsonDecoderModel: self.create_json_decoder,
 797            JsonItemsDecoderModel: self.create_json_items_decoder,
 798            JsonlDecoderModel: self.create_jsonl_decoder,
 799            JsonSchemaPropertySelectorModel: self.create_json_schema_property_selector,
 800            GzipDecoderModel: self.create_gzip_decoder,
 801            KeysToLowerModel: self.create_keys_to_lower_transformation,
 802            KeysToSnakeCaseModel: self.create_keys_to_snake_transformation,
 803            KeysReplaceModel: self.create_keys_replace_transformation,
 804            FlattenFieldsModel: self.create_flatten_fields,
 805            DpathFlattenFieldsModel: self.create_dpath_flatten_fields,
 806            IterableDecoderModel: self.create_iterable_decoder,
 807            XmlDecoderModel: self.create_xml_decoder,
 808            JsonFileSchemaLoaderModel: self.create_json_file_schema_loader,
 809            DynamicSchemaLoaderModel: self.create_dynamic_schema_loader,
 810            SchemaTypeIdentifierModel: self.create_schema_type_identifier,
 811            TypesMapModel: self.create_types_map,
 812            ComplexFieldTypeModel: self.create_complex_field_type,
 813            JwtAuthenticatorModel: self.create_jwt_authenticator,
 814            LegacyToPerPartitionStateMigrationModel: self.create_legacy_to_per_partition_state_migration,
 815            ListPartitionRouterModel: self.create_list_partition_router,
 816            MinMaxDatetimeModel: self.create_min_max_datetime,
 817            NoAuthModel: self.create_no_auth,
 818            NoPaginationModel: self.create_no_pagination,
 819            OAuthAuthenticatorModel: self.create_oauth_authenticator,
 820            OffsetIncrementModel: self.create_offset_increment,
 821            PageIncrementModel: self.create_page_increment,
 822            ParentStreamConfigModel: self.create_parent_stream_config_with_substream_wrapper,
 823            PredicateValidatorModel: self.create_predicate_validator,
 824            PropertiesFromEndpointModel: self.create_properties_from_endpoint,
 825            PropertyChunkingModel: self.create_property_chunking,
 826            QueryPropertiesModel: self.create_query_properties,
 827            RecordExpanderModel: self.create_record_expander,
 828            RecordFilterModel: self.create_record_filter,
 829            RecordSelectorModel: self.create_record_selector,
 830            RemoveFieldsModel: self.create_remove_fields,
 831            RequestPathModel: self.create_request_path,
 832            RequestOptionModel: self.create_request_option,
 833            LegacySessionTokenAuthenticatorModel: self.create_legacy_session_token_authenticator,
 834            SelectiveAuthenticatorModel: self.create_selective_authenticator,
 835            SimpleRetrieverModel: self.create_simple_retriever,
 836            StateDelegatingStreamModel: self.create_state_delegating_stream,
 837            SpecModel: self.create_spec,
 838            SubstreamPartitionRouterModel: self.create_substream_partition_router,
 839            ValidateAdheresToSchemaModel: self.create_validate_adheres_to_schema,
 840            WaitTimeFromHeaderModel: self.create_wait_time_from_header,
 841            WaitUntilTimeFromHeaderModel: self.create_wait_until_time_from_header,
 842            AsyncRetrieverModel: self.create_async_retriever,
 843            HttpComponentsResolverModel: self.create_http_components_resolver,
 844            ConfigComponentsResolverModel: self.create_config_components_resolver,
 845            ParametrizedComponentsResolverModel: self.create_parametrized_components_resolver,
 846            StreamConfigModel: self.create_stream_config,
 847            ComponentMappingDefinitionModel: self.create_components_mapping_definition,
 848            ZipfileDecoderModel: self.create_zipfile_decoder,
 849            HTTPAPIBudgetModel: self.create_http_api_budget,
 850            FileUploaderModel: self.create_file_uploader,
 851            FixedWindowCallRatePolicyModel: self.create_fixed_window_call_rate_policy,
 852            MovingWindowCallRatePolicyModel: self.create_moving_window_call_rate_policy,
 853            UnlimitedCallRatePolicyModel: self.create_unlimited_call_rate_policy,
 854            RateModel: self.create_rate,
 855            HttpRequestRegexMatcherModel: self.create_http_request_matcher,
 856            RateLimitedMultipleTokenAuthenticatorModel: self.create_rate_limited_multiple_token_authenticator,
 857            GroupingPartitionRouterModel: self.create_grouping_partition_router,
 858            UnionPartitionRouterModel: self.create_union_partition_router,
 859        }
 860
 861        # Needed for the case where we need to perform a second parse on the fields of a custom component
 862        self.TYPE_NAME_TO_MODEL = {cls.__name__: cls for cls in self.PYDANTIC_MODEL_TO_CONSTRUCTOR}
 863
 864    @staticmethod
 865    def _create_stream_name_to_configured_stream(
 866        configured_catalog: Optional[ConfiguredAirbyteCatalog],
 867    ) -> Mapping[str, ConfiguredAirbyteStream]:
 868        return (
 869            {stream.stream.name: stream for stream in configured_catalog.streams}
 870            if configured_catalog
 871            else {}
 872        )
 873
 874    def create_component(
 875        self,
 876        model_type: Type[BaseModel],
 877        component_definition: ComponentDefinition,
 878        config: Config,
 879        **kwargs: Any,
 880    ) -> Any:
 881        """
 882        Takes a given Pydantic model type and Mapping representing a component definition and creates a declarative component and
 883        subcomponents which will be used at runtime. This is done by first parsing the mapping into a Pydantic model and then creating
 884        creating declarative components from that model.
 885
 886        :param model_type: The type of declarative component that is being initialized
 887        :param component_definition: The mapping that represents a declarative component
 888        :param config: The connector config that is provided by the customer
 889        :return: The declarative component to be used at runtime
 890        """
 891
 892        component_type = component_definition.get("type")
 893        if component_definition.get("type") != model_type.__name__:
 894            raise ValueError(
 895                f"Expected manifest component of type {model_type.__name__}, but received {component_type} instead"
 896            )
 897
 898        declarative_component_model = model_type.parse_obj(component_definition)
 899
 900        if not isinstance(declarative_component_model, model_type):
 901            raise ValueError(
 902                f"Expected {model_type.__name__} component, but received {declarative_component_model.__class__.__name__}"
 903            )
 904
 905        return self._create_component_from_model(
 906            model=declarative_component_model, config=config, **kwargs
 907        )
 908
 909    def _create_component_from_model(self, model: BaseModel, config: Config, **kwargs: Any) -> Any:
 910        if model.__class__ not in self.PYDANTIC_MODEL_TO_CONSTRUCTOR:
 911            raise ValueError(
 912                f"{model.__class__} with attributes {model} is not a valid component type"
 913            )
 914        component_constructor = self.PYDANTIC_MODEL_TO_CONSTRUCTOR.get(model.__class__)
 915        if not component_constructor:
 916            raise ValueError(f"Could not find constructor for {model.__class__}")
 917
 918        # collect deprecation warnings for supported models.
 919        if isinstance(model, BaseModelWithDeprecations):
 920            self._collect_model_deprecations(model)
 921
 922        return component_constructor(model=model, config=config, **kwargs)
 923
 924    def get_model_deprecations(self) -> List[ConnectorBuilderLogMessage]:
 925        """
 926        Returns the deprecation warnings that were collected during the creation of components.
 927        """
 928        return self._collected_deprecation_logs
 929
 930    def _collect_model_deprecations(self, model: BaseModelWithDeprecations) -> None:
 931        """
 932        Collects deprecation logs from the given model and appends any new logs to the internal collection.
 933
 934        This method checks if the provided model has deprecation logs (identified by the presence of the DEPRECATION_LOGS_TAG attribute and a non-None `_deprecation_logs` property). It iterates through each deprecation log in the model and appends it to the `_collected_deprecation_logs` list if it has not already been collected, ensuring that duplicate logs are avoided.
 935
 936        Args:
 937            model (BaseModelWithDeprecations): The model instance from which to collect deprecation logs.
 938        """
 939        if hasattr(model, DEPRECATION_LOGS_TAG) and model._deprecation_logs is not None:
 940            for log in model._deprecation_logs:
 941                # avoid duplicates for deprecation logs observed.
 942                if log not in self._collected_deprecation_logs:
 943                    self._collected_deprecation_logs.append(log)
 944
 945    def create_config_migration(
 946        self, model: ConfigMigrationModel, config: Config
 947    ) -> ConfigMigration:
 948        transformations: List[ConfigTransformation] = [
 949            self._create_component_from_model(transformation, config)
 950            for transformation in model.transformations
 951        ]
 952
 953        return ConfigMigration(
 954            description=model.description,
 955            transformations=transformations,
 956        )
 957
 958    def create_config_add_fields(
 959        self, model: ConfigAddFieldsModel, config: Config, **kwargs: Any
 960    ) -> ConfigAddFields:
 961        fields = [self._create_component_from_model(field, config) for field in model.fields]
 962        return ConfigAddFields(
 963            fields=fields,
 964            condition=model.condition or "",
 965        )
 966
 967    @staticmethod
 968    def create_config_remove_fields(
 969        model: ConfigRemoveFieldsModel, config: Config, **kwargs: Any
 970    ) -> ConfigRemoveFields:
 971        return ConfigRemoveFields(
 972            field_pointers=model.field_pointers,
 973            condition=model.condition or "",
 974        )
 975
 976    @staticmethod
 977    def create_config_remap_field(
 978        model: ConfigRemapFieldModel, config: Config, **kwargs: Any
 979    ) -> ConfigRemapField:
 980        mapping = cast(Mapping[str, Any], model.map)
 981        return ConfigRemapField(
 982            map=mapping,
 983            field_path=model.field_path,
 984            config=config,
 985        )
 986
 987    def create_dpath_validator(self, model: DpathValidatorModel, config: Config) -> DpathValidator:
 988        strategy = self._create_component_from_model(model.validation_strategy, config)
 989
 990        return DpathValidator(
 991            field_path=model.field_path,
 992            strategy=strategy,
 993        )
 994
 995    def create_predicate_validator(
 996        self, model: PredicateValidatorModel, config: Config
 997    ) -> PredicateValidator:
 998        strategy = self._create_component_from_model(model.validation_strategy, config)
 999
1000        return PredicateValidator(
1001            value=model.value,
1002            strategy=strategy,
1003        )
1004
1005    @staticmethod
1006    def create_validate_adheres_to_schema(
1007        model: ValidateAdheresToSchemaModel, config: Config, **kwargs: Any
1008    ) -> ValidateAdheresToSchema:
1009        base_schema = cast(Mapping[str, Any], model.base_schema)
1010        return ValidateAdheresToSchema(
1011            schema=base_schema,
1012        )
1013
1014    @staticmethod
1015    def create_added_field_definition(
1016        model: AddedFieldDefinitionModel, config: Config, **kwargs: Any
1017    ) -> AddedFieldDefinition:
1018        interpolated_value = InterpolatedString.create(
1019            model.value, parameters=model.parameters or {}
1020        )
1021        return AddedFieldDefinition(
1022            path=model.path,
1023            value=interpolated_value,
1024            value_type=ModelToComponentFactory._json_schema_type_name_to_type(model.value_type),
1025            parameters=model.parameters or {},
1026        )
1027
1028    def create_add_fields(self, model: AddFieldsModel, config: Config, **kwargs: Any) -> AddFields:
1029        added_field_definitions = [
1030            self._create_component_from_model(
1031                model=added_field_definition_model,
1032                value_type=ModelToComponentFactory._json_schema_type_name_to_type(
1033                    added_field_definition_model.value_type
1034                ),
1035                config=config,
1036            )
1037            for added_field_definition_model in model.fields
1038        ]
1039        return AddFields(
1040            fields=added_field_definitions,
1041            condition=model.condition or "",
1042            parameters=model.parameters or {},
1043        )
1044
1045    def create_keys_to_lower_transformation(
1046        self, model: KeysToLowerModel, config: Config, **kwargs: Any
1047    ) -> KeysToLowerTransformation:
1048        return KeysToLowerTransformation()
1049
1050    def create_keys_to_snake_transformation(
1051        self, model: KeysToSnakeCaseModel, config: Config, **kwargs: Any
1052    ) -> KeysToSnakeCaseTransformation:
1053        return KeysToSnakeCaseTransformation()
1054
1055    def create_keys_replace_transformation(
1056        self, model: KeysReplaceModel, config: Config, **kwargs: Any
1057    ) -> KeysReplaceTransformation:
1058        return KeysReplaceTransformation(
1059            old=model.old, new=model.new, parameters=model.parameters or {}
1060        )
1061
1062    def create_flatten_fields(
1063        self, model: FlattenFieldsModel, config: Config, **kwargs: Any
1064    ) -> FlattenFields:
1065        return FlattenFields(
1066            flatten_lists=model.flatten_lists if model.flatten_lists is not None else True
1067        )
1068
1069    def create_dpath_flatten_fields(
1070        self, model: DpathFlattenFieldsModel, config: Config, **kwargs: Any
1071    ) -> DpathFlattenFields:
1072        model_field_path: List[Union[InterpolatedString, str]] = [x for x in model.field_path]
1073        key_transformation = (
1074            KeyTransformation(
1075                config=config,
1076                prefix=model.key_transformation.prefix,
1077                suffix=model.key_transformation.suffix,
1078                parameters=model.parameters or {},
1079            )
1080            if model.key_transformation is not None
1081            else None
1082        )
1083        return DpathFlattenFields(
1084            config=config,
1085            field_path=model_field_path,
1086            delete_origin_value=model.delete_origin_value
1087            if model.delete_origin_value is not None
1088            else False,
1089            replace_record=model.replace_record if model.replace_record is not None else False,
1090            key_transformation=key_transformation,
1091            parameters=model.parameters or {},
1092        )
1093
1094    @staticmethod
1095    def _json_schema_type_name_to_type(value_type: Optional[ValueType]) -> Optional[Type[Any]]:
1096        if not value_type:
1097            return None
1098        names_to_types = {
1099            ValueType.string: str,
1100            ValueType.number: float,
1101            ValueType.integer: int,
1102            ValueType.boolean: bool,
1103        }
1104        return names_to_types[value_type]
1105
1106    def create_api_key_authenticator(
1107        self,
1108        model: ApiKeyAuthenticatorModel,
1109        config: Config,
1110        token_provider: Optional[TokenProvider] = None,
1111        **kwargs: Any,
1112    ) -> ApiKeyAuthenticator:
1113        if model.inject_into is None and model.header is None:
1114            raise ValueError(
1115                "Expected either inject_into or header to be set for ApiKeyAuthenticator"
1116            )
1117
1118        if model.inject_into is not None and model.header is not None:
1119            raise ValueError(
1120                "inject_into and header cannot be set both for ApiKeyAuthenticator - remove the deprecated header option"
1121            )
1122
1123        if token_provider is not None and model.api_token != "":
1124            raise ValueError(
1125                "If token_provider is set, api_token is ignored and has to be set to empty string."
1126            )
1127
1128        request_option = (
1129            self._create_component_from_model(
1130                model.inject_into, config, parameters=model.parameters or {}
1131            )
1132            if model.inject_into
1133            else RequestOption(
1134                inject_into=RequestOptionType.header,
1135                field_name=model.header or "",
1136                parameters=model.parameters or {},
1137            )
1138        )
1139
1140        return ApiKeyAuthenticator(
1141            token_provider=(
1142                token_provider
1143                if token_provider is not None
1144                else InterpolatedStringTokenProvider(
1145                    api_token=model.api_token or "",
1146                    config=config,
1147                    parameters=model.parameters or {},
1148                )
1149            ),
1150            request_option=request_option,
1151            config=config,
1152            parameters=model.parameters or {},
1153        )
1154
1155    def create_legacy_to_per_partition_state_migration(
1156        self,
1157        model: LegacyToPerPartitionStateMigrationModel,
1158        config: Mapping[str, Any],
1159        declarative_stream: DeclarativeStreamModel,
1160    ) -> LegacyToPerPartitionStateMigration:
1161        retriever = declarative_stream.retriever
1162        if not isinstance(retriever, (SimpleRetrieverModel, AsyncRetrieverModel)):
1163            raise ValueError(
1164                f"LegacyToPerPartitionStateMigrations can only be applied on a DeclarativeStream with a SimpleRetriever or AsyncRetriever. Got {type(retriever)}"
1165            )
1166        partition_router = retriever.partition_router
1167        if not isinstance(
1168            partition_router,
1169            (
1170                SubstreamPartitionRouterModel,
1171                CustomPartitionRouterModel,
1172                UnionPartitionRouterModel,
1173            ),
1174        ):
1175            raise ValueError(
1176                f"LegacyToPerPartitionStateMigrations can only be applied on a SimpleRetriever with a SubstreamPartitionRouter, UnionPartitionRouter or CustomPartitionRouter. Got {type(partition_router)}"
1177            )
1178        if not isinstance(partition_router, UnionPartitionRouterModel) and not hasattr(
1179            partition_router, "parent_stream_configs"
1180        ):
1181            raise ValueError(
1182                "LegacyToPerPartitionStateMigrations can only be applied with a parent stream configuration."
1183            )
1184
1185        if not hasattr(declarative_stream, "incremental_sync"):
1186            raise ValueError(
1187                "LegacyToPerPartitionStateMigrations can only be applied with an incremental_sync configuration."
1188            )
1189
1190        return LegacyToPerPartitionStateMigration(
1191            partition_router,  # type: ignore # was already checked above
1192            declarative_stream.incremental_sync,  # type: ignore # was already checked. Migration can be applied only to incremental streams.
1193            config,
1194            declarative_stream.parameters,  # type: ignore # different type is expected here Mapping[str, Any], got Dict[str, Any]
1195        )
1196
1197    def create_session_token_authenticator(
1198        self, model: SessionTokenAuthenticatorModel, config: Config, name: str, **kwargs: Any
1199    ) -> Union[ApiKeyAuthenticator, BearerAuthenticator]:
1200        decoder = (
1201            self._create_component_from_model(model=model.decoder, config=config)
1202            if model.decoder
1203            else JsonDecoder(parameters={})
1204        )
1205        login_requester = self._create_component_from_model(
1206            model=model.login_requester,
1207            config=config,
1208            name=f"{name}_login_requester",
1209            decoder=decoder,
1210        )
1211        token_provider = SessionTokenProvider(
1212            login_requester=login_requester,
1213            session_token_path=model.session_token_path,
1214            expiration_duration=parse_duration(model.expiration_duration)
1215            if model.expiration_duration
1216            else None,
1217            parameters=model.parameters or {},
1218            message_repository=self._message_repository,
1219            decoder=decoder,
1220        )
1221        if model.request_authentication.type == "Bearer":
1222            return ModelToComponentFactory.create_bearer_authenticator(
1223                BearerAuthenticatorModel(type="BearerAuthenticator", api_token=""),  # type: ignore # $parameters has a default value
1224                config,
1225                token_provider=token_provider,
1226            )
1227        else:
1228            # Get the api_token template if specified, default to just the session token
1229            api_token_template = (
1230                getattr(model.request_authentication, "api_token", None) or "{{ session_token }}"
1231            )
1232            final_token_provider: TokenProvider = InterpolatedSessionTokenProvider(
1233                config=config,
1234                api_token=api_token_template,
1235                session_token_provider=token_provider,
1236                parameters=model.parameters or {},
1237            )
1238            return self.create_api_key_authenticator(
1239                ApiKeyAuthenticatorModel(
1240                    type="ApiKeyAuthenticator",
1241                    api_token="",
1242                    inject_into=model.request_authentication.inject_into,
1243                ),  # type: ignore # $parameters and headers default to None
1244                config=config,
1245                token_provider=final_token_provider,
1246            )
1247
1248    @staticmethod
1249    def create_basic_http_authenticator(
1250        model: BasicHttpAuthenticatorModel, config: Config, **kwargs: Any
1251    ) -> BasicHttpAuthenticator:
1252        return BasicHttpAuthenticator(
1253            password=model.password or "",
1254            username=model.username,
1255            config=config,
1256            parameters=model.parameters or {},
1257        )
1258
1259    @staticmethod
1260    def create_bearer_authenticator(
1261        model: BearerAuthenticatorModel,
1262        config: Config,
1263        token_provider: Optional[TokenProvider] = None,
1264        **kwargs: Any,
1265    ) -> BearerAuthenticator:
1266        if token_provider is not None and model.api_token != "":
1267            raise ValueError(
1268                "If token_provider is set, api_token is ignored and has to be set to empty string."
1269            )
1270        return BearerAuthenticator(
1271            token_provider=(
1272                token_provider
1273                if token_provider is not None
1274                else InterpolatedStringTokenProvider(
1275                    api_token=model.api_token or "",
1276                    config=config,
1277                    parameters=model.parameters or {},
1278                )
1279            ),
1280            config=config,
1281            parameters=model.parameters or {},
1282        )
1283
1284    @staticmethod
1285    def create_dynamic_stream_check_config(
1286        model: DynamicStreamCheckConfigModel, config: Config, **kwargs: Any
1287    ) -> DynamicStreamCheckConfig:
1288        return DynamicStreamCheckConfig(
1289            dynamic_stream_name=model.dynamic_stream_name,
1290            stream_count=model.stream_count,
1291        )
1292
1293    def create_check_stream(
1294        self, model: CheckStreamModel, config: Config, **kwargs: Any
1295    ) -> CheckStream:
1296        if model.dynamic_streams_check_configs is None and model.stream_names is None:
1297            raise ValueError(
1298                "Expected either stream_names or dynamic_streams_check_configs to be set for CheckStream"
1299            )
1300
1301        dynamic_streams_check_configs = (
1302            [
1303                self._create_component_from_model(model=dynamic_stream_check_config, config=config)
1304                for dynamic_stream_check_config in model.dynamic_streams_check_configs
1305            ]
1306            if model.dynamic_streams_check_configs
1307            else []
1308        )
1309
1310        # `model.config_overrides` is deliberately not read here. The source applies it around the whole
1311        # check operation (`ConcurrentDeclarativeSource._config_overridden_for_check`), which is what makes
1312        # it work for every checker type rather than only this one. Do not wire it in a second time.
1313        return CheckStream(
1314            stream_names=model.stream_names or [],
1315            dynamic_streams_check_configs=dynamic_streams_check_configs,
1316            parameters={},
1317        )
1318
1319    @staticmethod
1320    def create_check_dynamic_stream(
1321        model: CheckDynamicStreamModel, config: Config, **kwargs: Any
1322    ) -> CheckDynamicStream:
1323        assert model.use_check_availability is not None  # for mypy
1324
1325        use_check_availability = model.use_check_availability
1326
1327        # See `create_check_stream`: `model.config_overrides` is applied by the source, not here.
1328        return CheckDynamicStream(
1329            stream_count=model.stream_count,
1330            use_check_availability=use_check_availability,
1331            parameters={},
1332        )
1333
1334    def create_composite_error_handler(
1335        self, model: CompositeErrorHandlerModel, config: Config, **kwargs: Any
1336    ) -> CompositeErrorHandler:
1337        error_handlers = [
1338            self._create_component_from_model(model=error_handler_model, config=config)
1339            for error_handler_model in model.error_handlers
1340        ]
1341        return CompositeErrorHandler(
1342            error_handlers=error_handlers, parameters=model.parameters or {}
1343        )
1344
1345    @staticmethod
1346    def create_concurrency_level(
1347        model: ConcurrencyLevelModel, config: Config, **kwargs: Any
1348    ) -> ConcurrencyLevel:
1349        return ConcurrencyLevel(
1350            default_concurrency=model.default_concurrency,
1351            max_concurrency=model.max_concurrency,
1352            config=config,
1353            parameters={},
1354        )
1355
1356    @staticmethod
1357    def apply_stream_state_migrations(
1358        stream_state_migrations: List[Any] | None, stream_state: MutableMapping[str, Any]
1359    ) -> MutableMapping[str, Any]:
1360        if stream_state_migrations:
1361            for state_migration in stream_state_migrations:
1362                if state_migration.should_migrate(stream_state):
1363                    # The state variable is expected to be mutable but the migrate method returns an immutable mapping.
1364                    stream_state = dict(state_migration.migrate(stream_state))
1365        return stream_state
1366
1367    def create_concurrent_cursor_from_datetime_based_cursor(
1368        self,
1369        model_type: Type[BaseModel],
1370        component_definition: ComponentDefinition,
1371        stream_name: str,
1372        stream_namespace: Optional[str],
1373        stream_state: MutableMapping[str, Any],
1374        config: Config,
1375        message_repository: Optional[MessageRepository] = None,
1376        runtime_lookback_window: Optional[datetime.timedelta] = None,
1377        **kwargs: Any,
1378    ) -> ConcurrentCursor:
1379        component_type = component_definition.get("type")
1380        if component_definition.get("type") != model_type.__name__:
1381            raise ValueError(
1382                f"Expected manifest component of type {model_type.__name__}, but received {component_type} instead"
1383            )
1384
1385        # FIXME the interfaces of the concurrent cursor are kind of annoying as they take a `ComponentDefinition` instead of the actual model. This was done because the ConcurrentDeclarativeSource didn't have access to the models [here for example](https://github.com/airbytehq/airbyte-python-cdk/blob/f525803b3fec9329e4cc8478996a92bf884bfde9/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L354C54-L354C91). So now we have two cases:
1386        # * The ComponentDefinition comes from model.__dict__ in which case we have `parameters`
1387        # * The ComponentDefinition comes from the manifest as a dict in which case we have `$parameters`
1388        # We should change those interfaces to use the model once we clean up the code in CDS at which point the parameter propagation should happen as part of the ModelToComponentFactory.
1389        if "$parameters" not in component_definition and "parameters" in component_definition:
1390            component_definition["$parameters"] = component_definition.get("parameters")  # type: ignore  # This is a dict
1391        datetime_based_cursor_model = model_type.parse_obj(component_definition)
1392
1393        if not isinstance(datetime_based_cursor_model, DatetimeBasedCursorModel):
1394            raise ValueError(
1395                f"Expected {model_type.__name__} component, but received {datetime_based_cursor_model.__class__.__name__}"
1396            )
1397
1398        model_parameters = datetime_based_cursor_model.parameters or {}
1399
1400        cursor_field = self._get_catalog_defined_cursor_field(
1401            stream_name=stream_name,
1402            allow_catalog_defined_cursor_field=datetime_based_cursor_model.allow_catalog_defined_cursor_field
1403            or False,
1404        )
1405
1406        if not cursor_field:
1407            interpolated_cursor_field = InterpolatedString.create(
1408                datetime_based_cursor_model.cursor_field,
1409                parameters=model_parameters,
1410            )
1411            cursor_field = CursorField(
1412                cursor_field_key=interpolated_cursor_field.eval(config=config),
1413                supports_catalog_defined_cursor_field=datetime_based_cursor_model.allow_catalog_defined_cursor_field
1414                or False,
1415            )
1416
1417        interpolated_partition_field_start = InterpolatedString.create(
1418            datetime_based_cursor_model.partition_field_start or "start_time",
1419            parameters=model_parameters,
1420        )
1421        interpolated_partition_field_end = InterpolatedString.create(
1422            datetime_based_cursor_model.partition_field_end or "end_time",
1423            parameters=model_parameters,
1424        )
1425
1426        slice_boundary_fields = (
1427            interpolated_partition_field_start.eval(config=config),
1428            interpolated_partition_field_end.eval(config=config),
1429        )
1430
1431        datetime_format = datetime_based_cursor_model.datetime_format
1432
1433        cursor_granularity = (
1434            parse_duration(datetime_based_cursor_model.cursor_granularity)
1435            if datetime_based_cursor_model.cursor_granularity
1436            else None
1437        )
1438
1439        lookback_window = None
1440        interpolated_lookback_window = (
1441            InterpolatedString.create(
1442                datetime_based_cursor_model.lookback_window,
1443                parameters=model_parameters,
1444            )
1445            if datetime_based_cursor_model.lookback_window
1446            else None
1447        )
1448        if interpolated_lookback_window:
1449            evaluated_lookback_window = interpolated_lookback_window.eval(config=config)
1450            if evaluated_lookback_window:
1451                lookback_window = parse_duration(evaluated_lookback_window)
1452
1453        connector_state_converter: DateTimeStreamStateConverter
1454        connector_state_converter = CustomFormatConcurrentStreamStateConverter(
1455            datetime_format=datetime_format,
1456            input_datetime_formats=datetime_based_cursor_model.cursor_datetime_formats,
1457            is_sequential_state=True,  # ConcurrentPerPartitionCursor only works with sequential state
1458            cursor_granularity=cursor_granularity,
1459        )
1460
1461        # Adjusts the stream state by applying the runtime lookback window.
1462        # This is used to ensure correct state handling in case of failed partitions.
1463        stream_state_value = stream_state.get(cursor_field.cursor_field_key)
1464        if runtime_lookback_window and stream_state_value:
1465            new_stream_state = (
1466                connector_state_converter.parse_timestamp(stream_state_value)
1467                - runtime_lookback_window
1468            )
1469            stream_state[cursor_field.cursor_field_key] = connector_state_converter.output_format(
1470                new_stream_state
1471            )
1472
1473        start_date_runtime_value: Union[InterpolatedString, str, MinMaxDatetime]
1474        if isinstance(datetime_based_cursor_model.start_datetime, MinMaxDatetimeModel):
1475            start_date_runtime_value = self.create_min_max_datetime(
1476                model=datetime_based_cursor_model.start_datetime, config=config
1477            )
1478        else:
1479            start_date_runtime_value = datetime_based_cursor_model.start_datetime
1480
1481        end_date_runtime_value: Optional[Union[InterpolatedString, str, MinMaxDatetime]]
1482        if isinstance(datetime_based_cursor_model.end_datetime, MinMaxDatetimeModel):
1483            end_date_runtime_value = self.create_min_max_datetime(
1484                model=datetime_based_cursor_model.end_datetime, config=config
1485            )
1486        else:
1487            end_date_runtime_value = datetime_based_cursor_model.end_datetime
1488
1489        interpolated_start_date = MinMaxDatetime.create(
1490            interpolated_string_or_min_max_datetime=start_date_runtime_value,
1491            parameters=datetime_based_cursor_model.parameters,
1492        )
1493        interpolated_end_date = (
1494            None
1495            if not end_date_runtime_value
1496            else MinMaxDatetime.create(
1497                end_date_runtime_value, datetime_based_cursor_model.parameters
1498            )
1499        )
1500
1501        # If datetime format is not specified then start/end datetime should inherit it from the stream slicer
1502        if not interpolated_start_date.datetime_format:
1503            interpolated_start_date.datetime_format = datetime_format
1504        if interpolated_end_date and not interpolated_end_date.datetime_format:
1505            interpolated_end_date.datetime_format = datetime_format
1506
1507        start_date = interpolated_start_date.get_datetime(config=config)
1508        end_date_provider = (
1509            partial(interpolated_end_date.get_datetime, config)
1510            if interpolated_end_date
1511            else connector_state_converter.get_end_provider()
1512        )
1513
1514        if (
1515            datetime_based_cursor_model.step and not datetime_based_cursor_model.cursor_granularity
1516        ) or (
1517            not datetime_based_cursor_model.step and datetime_based_cursor_model.cursor_granularity
1518        ):
1519            raise ValueError(
1520                f"If step is defined, cursor_granularity should be as well and vice-versa. "
1521                f"Right now, step is `{datetime_based_cursor_model.step}` and cursor_granularity is `{datetime_based_cursor_model.cursor_granularity}`"
1522            )
1523
1524        # When step is not defined, default to a step size from the starting date to the present moment
1525        step_length = datetime.timedelta.max
1526        interpolated_step = (
1527            InterpolatedString.create(
1528                datetime_based_cursor_model.step,
1529                parameters=model_parameters,
1530            )
1531            if datetime_based_cursor_model.step
1532            else None
1533        )
1534        if interpolated_step:
1535            evaluated_step = interpolated_step.eval(config)
1536            if evaluated_step:
1537                step_length = parse_duration(evaluated_step)
1538
1539        clamping_strategy: ClampingStrategy = NoClamping()
1540        if datetime_based_cursor_model.clamping:
1541            # While it is undesirable to interpolate within the model factory (as opposed to at runtime),
1542            # it is still better than shifting interpolation low-code concept into the ConcurrentCursor runtime
1543            # object which we want to keep agnostic of being low-code
1544            target = InterpolatedString(
1545                string=datetime_based_cursor_model.clamping.target,
1546                parameters=model_parameters,
1547            )
1548            evaluated_target = target.eval(config=config)
1549            match evaluated_target:
1550                case "DAY":
1551                    clamping_strategy = DayClampingStrategy()
1552                    end_date_provider = ClampingEndProvider(
1553                        DayClampingStrategy(is_ceiling=False),
1554                        end_date_provider,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1555                        granularity=cursor_granularity or datetime.timedelta(seconds=1),
1556                    )
1557                case "WEEK":
1558                    if (
1559                        not datetime_based_cursor_model.clamping.target_details
1560                        or "weekday" not in datetime_based_cursor_model.clamping.target_details
1561                    ):
1562                        raise ValueError(
1563                            "Given WEEK clamping, weekday needs to be provided as target_details"
1564                        )
1565                    weekday = self._assemble_weekday(
1566                        datetime_based_cursor_model.clamping.target_details["weekday"]
1567                    )
1568                    clamping_strategy = WeekClampingStrategy(weekday)
1569                    end_date_provider = ClampingEndProvider(
1570                        WeekClampingStrategy(weekday, is_ceiling=False),
1571                        end_date_provider,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1572                        granularity=cursor_granularity or datetime.timedelta(days=1),
1573                    )
1574                case "MONTH":
1575                    clamping_strategy = MonthClampingStrategy()
1576                    end_date_provider = ClampingEndProvider(
1577                        MonthClampingStrategy(is_ceiling=False),
1578                        end_date_provider,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1579                        granularity=cursor_granularity or datetime.timedelta(days=1),
1580                    )
1581                case _:
1582                    raise ValueError(
1583                        f"Invalid clamping target {evaluated_target}, expected DAY, WEEK, MONTH"
1584                    )
1585
1586        return ConcurrentCursor(
1587            stream_name=stream_name,
1588            stream_namespace=stream_namespace,
1589            stream_state=stream_state,
1590            message_repository=message_repository or self._message_repository,
1591            connector_state_manager=self._connector_state_manager,
1592            connector_state_converter=connector_state_converter,
1593            cursor_field=cursor_field,
1594            slice_boundary_fields=slice_boundary_fields,
1595            start=start_date,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1596            end_provider=end_date_provider,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1597            lookback_window=lookback_window,
1598            slice_range=step_length,
1599            cursor_granularity=cursor_granularity,
1600            clamping_strategy=clamping_strategy,
1601        )
1602
1603    def create_concurrent_cursor_from_incrementing_count_cursor(
1604        self,
1605        model_type: Type[BaseModel],
1606        component_definition: ComponentDefinition,
1607        stream_name: str,
1608        stream_namespace: Optional[str],
1609        stream_state: MutableMapping[str, Any],
1610        config: Config,
1611        message_repository: Optional[MessageRepository] = None,
1612        **kwargs: Any,
1613    ) -> ConcurrentCursor:
1614        component_type = component_definition.get("type")
1615        if component_definition.get("type") != model_type.__name__:
1616            raise ValueError(
1617                f"Expected manifest component of type {model_type.__name__}, but received {component_type} instead"
1618            )
1619
1620        incrementing_count_cursor_model = model_type.parse_obj(component_definition)
1621
1622        if not isinstance(incrementing_count_cursor_model, IncrementingCountCursorModel):
1623            raise ValueError(
1624                f"Expected {model_type.__name__} component, but received {incrementing_count_cursor_model.__class__.__name__}"
1625            )
1626
1627        start_value: Union[int, str, None] = incrementing_count_cursor_model.start_value
1628        # Pydantic Union type coercion can convert int 0 to string '0' depending on Union order.
1629        # We need to handle both int and str representations of numeric values.
1630        # Evaluate the InterpolatedString and convert to int for the ConcurrentCursor.
1631        if start_value is not None:
1632            interpolated_start_value = InterpolatedString.create(
1633                str(start_value),  # Ensure we pass a string to InterpolatedString.create
1634                parameters=incrementing_count_cursor_model.parameters or {},
1635            )
1636            evaluated_start_value: int = int(interpolated_start_value.eval(config=config))
1637        else:
1638            evaluated_start_value = 0
1639
1640        cursor_field = self._get_catalog_defined_cursor_field(
1641            stream_name=stream_name,
1642            allow_catalog_defined_cursor_field=incrementing_count_cursor_model.allow_catalog_defined_cursor_field
1643            or False,
1644        )
1645
1646        if not cursor_field:
1647            interpolated_cursor_field = InterpolatedString.create(
1648                incrementing_count_cursor_model.cursor_field,
1649                parameters=incrementing_count_cursor_model.parameters or {},
1650            )
1651            cursor_field = CursorField(
1652                cursor_field_key=interpolated_cursor_field.eval(config=config),
1653                supports_catalog_defined_cursor_field=incrementing_count_cursor_model.allow_catalog_defined_cursor_field
1654                or False,
1655            )
1656
1657        connector_state_converter = IncrementingCountStreamStateConverter(
1658            is_sequential_state=True,  # ConcurrentPerPartitionCursor only works with sequential state
1659        )
1660
1661        return ConcurrentCursor(
1662            stream_name=stream_name,
1663            stream_namespace=stream_namespace,
1664            stream_state=stream_state,
1665            message_repository=message_repository or self._message_repository,
1666            connector_state_manager=self._connector_state_manager,
1667            connector_state_converter=connector_state_converter,
1668            cursor_field=cursor_field,
1669            slice_boundary_fields=None,
1670            start=evaluated_start_value,  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1671            end_provider=connector_state_converter.get_end_provider(),  # type: ignore  # Having issues w/ inspection for GapType and CursorValueType as shown in existing tests. Confirmed functionality is working in practice
1672        )
1673
1674    def _assemble_weekday(self, weekday: str) -> Weekday:
1675        match weekday:
1676            case "MONDAY":
1677                return Weekday.MONDAY
1678            case "TUESDAY":
1679                return Weekday.TUESDAY
1680            case "WEDNESDAY":
1681                return Weekday.WEDNESDAY
1682            case "THURSDAY":
1683                return Weekday.THURSDAY
1684            case "FRIDAY":
1685                return Weekday.FRIDAY
1686            case "SATURDAY":
1687                return Weekday.SATURDAY
1688            case "SUNDAY":
1689                return Weekday.SUNDAY
1690            case _:
1691                raise ValueError(f"Unknown weekday {weekday}")
1692
1693    def create_concurrent_cursor_from_perpartition_cursor(
1694        self,
1695        state_manager: ConnectorStateManager,
1696        model_type: Type[BaseModel],
1697        component_definition: ComponentDefinition,
1698        stream_name: str,
1699        stream_namespace: Optional[str],
1700        config: Config,
1701        stream_state: MutableMapping[str, Any],
1702        partition_router: PartitionRouter,
1703        attempt_to_create_cursor_if_not_provided: bool = False,
1704        **kwargs: Any,
1705    ) -> ConcurrentPerPartitionCursor:
1706        component_type = component_definition.get("type")
1707        if component_definition.get("type") != model_type.__name__:
1708            raise ValueError(
1709                f"Expected manifest component of type {model_type.__name__}, but received {component_type} instead"
1710            )
1711
1712        # FIXME the interfaces of the concurrent cursor are kind of annoying as they take a `ComponentDefinition` instead of the actual model. This was done because the ConcurrentDeclarativeSource didn't have access to the models [here for example](https://github.com/airbytehq/airbyte-python-cdk/blob/f525803b3fec9329e4cc8478996a92bf884bfde9/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L354C54-L354C91). So now we have two cases:
1713        # * The ComponentDefinition comes from model.__dict__ in which case we have `parameters`
1714        # * The ComponentDefinition comes from the manifest as a dict in which case we have `$parameters`
1715        # We should change those interfaces to use the model once we clean up the code in CDS at which point the parameter propagation should happen as part of the ModelToComponentFactory.
1716        if "$parameters" not in component_definition and "parameters" in component_definition:
1717            component_definition["$parameters"] = component_definition.get("parameters")  # type: ignore  # This is a dict
1718        datetime_based_cursor_model = model_type.parse_obj(component_definition)
1719
1720        if not isinstance(datetime_based_cursor_model, DatetimeBasedCursorModel):
1721            raise ValueError(
1722                f"Expected {model_type.__name__} component, but received {datetime_based_cursor_model.__class__.__name__}"
1723            )
1724
1725        cursor_field = self._get_catalog_defined_cursor_field(
1726            stream_name=stream_name,
1727            allow_catalog_defined_cursor_field=datetime_based_cursor_model.allow_catalog_defined_cursor_field
1728            or False,
1729        )
1730
1731        if not cursor_field:
1732            interpolated_cursor_field = InterpolatedString.create(
1733                datetime_based_cursor_model.cursor_field,
1734                # FIXME the interfaces of the concurrent cursor are kind of annoying as they take a `ComponentDefinition` instead of the actual model. This was done because the ConcurrentDeclarativeSource didn't have access to the models [here for example](https://github.com/airbytehq/airbyte-python-cdk/blob/f525803b3fec9329e4cc8478996a92bf884bfde9/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L354C54-L354C91). So now we have two cases:
1735                # * The ComponentDefinition comes from model.__dict__ in which case we have `parameters`
1736                # * The ComponentDefinition comes from the manifest as a dict in which case we have `$parameters`
1737                # We should change those interfaces to use the model once we clean up the code in CDS at which point the parameter propagation should happen as part of the ModelToComponentFactory.
1738                parameters=datetime_based_cursor_model.parameters or {},
1739            )
1740            cursor_field = CursorField(
1741                cursor_field_key=interpolated_cursor_field.eval(config=config),
1742                supports_catalog_defined_cursor_field=datetime_based_cursor_model.allow_catalog_defined_cursor_field
1743                or False,
1744            )
1745
1746        datetime_format = datetime_based_cursor_model.datetime_format
1747
1748        cursor_granularity = (
1749            parse_duration(datetime_based_cursor_model.cursor_granularity)
1750            if datetime_based_cursor_model.cursor_granularity
1751            else None
1752        )
1753
1754        connector_state_converter: DateTimeStreamStateConverter
1755        connector_state_converter = CustomFormatConcurrentStreamStateConverter(
1756            datetime_format=datetime_format,
1757            input_datetime_formats=datetime_based_cursor_model.cursor_datetime_formats,
1758            is_sequential_state=True,  # ConcurrentPerPartitionCursor only works with sequential state
1759            cursor_granularity=cursor_granularity,
1760        )
1761
1762        # Create the cursor factory
1763        cursor_factory = ConcurrentCursorFactory(
1764            partial(
1765                self.create_concurrent_cursor_from_datetime_based_cursor,
1766                state_manager=state_manager,
1767                model_type=model_type,
1768                component_definition=component_definition,
1769                stream_name=stream_name,
1770                stream_namespace=stream_namespace,
1771                config=config,
1772                message_repository=NoopMessageRepository(),
1773            )
1774        )
1775
1776        # Per-partition state doesn't make sense for GroupingPartitionRouter, so force the global state
1777        use_global_cursor = isinstance(
1778            partition_router, GroupingPartitionRouter
1779        ) or component_definition.get("global_substream_cursor", False)
1780
1781        # Return the concurrent cursor and state converter
1782        return ConcurrentPerPartitionCursor(
1783            cursor_factory=cursor_factory,
1784            partition_router=partition_router,
1785            stream_name=stream_name,
1786            stream_namespace=stream_namespace,
1787            stream_state=stream_state,
1788            message_repository=self._message_repository,  # type: ignore
1789            connector_state_manager=state_manager,
1790            connector_state_converter=connector_state_converter,
1791            cursor_field=cursor_field,
1792            use_global_cursor=use_global_cursor,
1793            attempt_to_create_cursor_if_not_provided=attempt_to_create_cursor_if_not_provided,
1794        )
1795
1796    @staticmethod
1797    def create_constant_backoff_strategy(
1798        model: ConstantBackoffStrategyModel, config: Config, **kwargs: Any
1799    ) -> ConstantBackoffStrategy:
1800        ModelToComponentFactory._validate_jitter_range(model.jitter_range_in_seconds)
1801        return ConstantBackoffStrategy(
1802            backoff_time_in_seconds=model.backoff_time_in_seconds,
1803            jitter_range_in_seconds=model.jitter_range_in_seconds,
1804            config=config,
1805            parameters=model.parameters or {},
1806        )
1807
1808    @staticmethod
1809    def _validate_jitter_range(jitter_range_in_seconds: Optional[float]) -> None:
1810        if jitter_range_in_seconds is not None and jitter_range_in_seconds < 0:
1811            raise ValueError("jitter_range_in_seconds must be greater than or equal to 0")
1812
1813    def create_cursor_pagination(
1814        self, model: CursorPaginationModel, config: Config, decoder: Decoder, **kwargs: Any
1815    ) -> CursorPaginationStrategy:
1816        if isinstance(decoder, PaginationDecoderDecorator):
1817            inner_decoder = decoder.decoder
1818        else:
1819            inner_decoder = decoder
1820            decoder = PaginationDecoderDecorator(decoder=decoder)
1821
1822        if self._is_supported_decoder_for_pagination(inner_decoder):
1823            decoder_to_use = decoder
1824        else:
1825            raise ValueError(
1826                self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(inner_decoder))
1827            )
1828
1829        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
1830        # If page_size is a string that represents an integer (not an interpolation), convert it back.
1831        page_size = model.page_size
1832        if isinstance(page_size, str) and page_size.isdigit():
1833            page_size = int(page_size)
1834
1835        return CursorPaginationStrategy(
1836            cursor_value=model.cursor_value,
1837            decoder=decoder_to_use,
1838            page_size=page_size,
1839            stop_condition=model.stop_condition,
1840            config=config,
1841            parameters=model.parameters or {},
1842        )
1843
1844    def create_custom_component(self, model: Any, config: Config, **kwargs: Any) -> Any:
1845        """
1846        Generically creates a custom component based on the model type and a class_name reference to the custom Python class being
1847        instantiated. Only the model's additional properties that match the custom class definition are passed to the constructor
1848        :param model: The Pydantic model of the custom component being created
1849        :param config: The custom defined connector config
1850        :return: The declarative component built from the Pydantic model to be used at runtime
1851        """
1852        # Instantiating a custom component means importing and executing arbitrary code referenced
1853        # by `class_name`. Manifests supplied by a caller, whether through the config or directly to
1854        # the manifest server, are untrusted input and could point `class_name` at any importable
1855        # callable, so they honor the same `AIRBYTE_ENABLE_UNSAFE_CODE` gate as injected
1856        # `components.py` code. Manifests bundled in a published connector image are trusted and may
1857        # always use their bundled custom components.
1858        manifest_is_untrusted = not self._custom_components_trusted or bool(
1859            config.get(INJECTED_MANIFEST)
1860        )
1861        if manifest_is_untrusted and not custom_code_execution_permitted():
1862            raise AirbyteCustomCodeNotPermittedError
1863
1864        custom_component_class = self._get_class_from_fully_qualified_class_name(model.class_name)
1865        component_fields = get_type_hints(custom_component_class)
1866        model_args = model.dict()
1867        model_args["config"] = config
1868
1869        # There are cases where a parent component will pass arguments to a child component via kwargs. When there are field collisions
1870        # we defer to these arguments over the component's definition
1871        for key, arg in kwargs.items():
1872            model_args[key] = arg
1873
1874        # Pydantic is unable to parse a custom component's fields that are subcomponents into models because their fields and types are not
1875        # defined in the schema. The fields and types are defined within the Python class implementation. Pydantic can only parse down to
1876        # the custom component and this code performs a second parse to convert the sub-fields first into models, then declarative components
1877        for model_field, model_value in model_args.items():
1878            # If a custom component field doesn't have a type set, we try to use the type hints to infer the type
1879            if (
1880                isinstance(model_value, dict)
1881                and "type" not in model_value
1882                and model_field in component_fields
1883            ):
1884                derived_type = self._derive_component_type_from_type_hints(
1885                    component_fields.get(model_field)
1886                )
1887                if derived_type:
1888                    model_value["type"] = derived_type
1889
1890            if self._is_component(model_value):
1891                model_args[model_field] = self._create_nested_component(
1892                    model,
1893                    model_field,
1894                    model_value,
1895                    config,
1896                    **kwargs,
1897                )
1898            elif isinstance(model_value, list):
1899                vals = []
1900                for v in model_value:
1901                    if isinstance(v, dict) and "type" not in v and model_field in component_fields:
1902                        derived_type = self._derive_component_type_from_type_hints(
1903                            component_fields.get(model_field)
1904                        )
1905                        if derived_type:
1906                            v["type"] = derived_type
1907                    if self._is_component(v):
1908                        vals.append(
1909                            self._create_nested_component(
1910                                model,
1911                                model_field,
1912                                v,
1913                                config,
1914                                **kwargs,
1915                            )
1916                        )
1917                    else:
1918                        vals.append(v)
1919                model_args[model_field] = vals
1920
1921        kwargs = {
1922            class_field: model_args[class_field]
1923            for class_field in component_fields.keys()
1924            if class_field in model_args
1925        }
1926
1927        if "api_budget" in component_fields and kwargs.get("api_budget") is None:
1928            kwargs["api_budget"] = self._api_budget
1929
1930        return custom_component_class(**kwargs)
1931
1932    @staticmethod
1933    def _get_class_from_fully_qualified_class_name(
1934        full_qualified_class_name: str,
1935    ) -> Any:
1936        """Get a class from its fully qualified name.
1937
1938        If a custom components module is needed, we assume it is already registered - probably
1939        as `source_declarative_manifest.components` or `components`.
1940
1941        Args:
1942            full_qualified_class_name (str): The fully qualified name of the class (e.g., "module.ClassName").
1943
1944        Returns:
1945            Any: The class object.
1946
1947        Raises:
1948            ValueError: If the class cannot be loaded.
1949        """
1950        split = full_qualified_class_name.split(".")
1951        module_name_full = ".".join(split[:-1])
1952        class_name = split[-1]
1953
1954        try:
1955            module_ref = importlib.import_module(module_name_full)
1956        except ModuleNotFoundError as e:
1957            if split[0] == "source_declarative_manifest":
1958                # During testing, the modules containing the custom components are not moved to source_declarative_manifest. In order to run the test, add the source folder to your PYTHONPATH or add it runtime using sys.path.append
1959                try:
1960                    import os
1961
1962                    module_name_with_source_declarative_manifest = ".".join(split[1:-1])
1963                    module_ref = importlib.import_module(
1964                        module_name_with_source_declarative_manifest
1965                    )
1966                except ModuleNotFoundError:
1967                    raise ValueError(f"Could not load module `{module_name_full}`.") from e
1968            else:
1969                raise ValueError(f"Could not load module `{module_name_full}`.") from e
1970
1971        try:
1972            return getattr(module_ref, class_name)
1973        except AttributeError as e:
1974            raise ValueError(
1975                f"Could not load class `{class_name}` from module `{module_name_full}`.",
1976            ) from e
1977
1978    @staticmethod
1979    def _derive_component_type_from_type_hints(field_type: Any) -> Optional[str]:
1980        interface = field_type
1981        while True:
1982            origin = get_origin(interface)
1983            if origin:
1984                # Unnest types until we reach the raw type
1985                # List[T] -> T
1986                # Optional[List[T]] -> T
1987                args = get_args(interface)
1988                interface = args[0]
1989            else:
1990                break
1991        if isinstance(interface, type) and not ModelToComponentFactory.is_builtin_type(interface):
1992            return interface.__name__
1993        return None
1994
1995    @staticmethod
1996    def is_builtin_type(cls: Optional[Type[Any]]) -> bool:
1997        if not cls:
1998            return False
1999        return cls.__module__ == "builtins"
2000
2001    @staticmethod
2002    def _extract_missing_parameters(error: TypeError) -> List[str]:
2003        parameter_search = re.search(r"keyword-only.*:\s(.*)", str(error))
2004        if parameter_search:
2005            return re.findall(r"\'(.+?)\'", parameter_search.group(1))
2006        else:
2007            return []
2008
2009    def _create_nested_component(
2010        self, model: Any, model_field: str, model_value: Any, config: Config, **kwargs: Any
2011    ) -> Any:
2012        type_name = model_value.get("type", None)
2013        if not type_name:
2014            # If no type is specified, we can assume this is a dictionary object which can be returned instead of a subcomponent
2015            return model_value
2016
2017        model_type = self.TYPE_NAME_TO_MODEL.get(type_name, None)
2018        if model_type:
2019            parsed_model = model_type.parse_obj(model_value)
2020            try:
2021                # To improve usability of the language, certain fields are shared between components. This can come in the form of
2022                # a parent component passing some of its fields to a child component or the parent extracting fields from other child
2023                # components and passing it to others. One example is the DefaultPaginator referencing the HttpRequester url_base
2024                # while constructing a SimpleRetriever. However, custom components don't support this behavior because they are created
2025                # generically in create_custom_component(). This block allows developers to specify extra arguments in $parameters that
2026                # are needed by a component and could not be shared.
2027                model_constructor = self.PYDANTIC_MODEL_TO_CONSTRUCTOR.get(parsed_model.__class__)
2028                constructor_kwargs = inspect.getfullargspec(model_constructor).kwonlyargs
2029                model_parameters = model_value.get("$parameters", {})
2030                matching_parameters = {
2031                    kwarg: model_parameters[kwarg]
2032                    for kwarg in constructor_kwargs
2033                    if kwarg in model_parameters
2034                }
2035                matching_kwargs = {
2036                    kwarg: kwargs[kwarg] for kwarg in constructor_kwargs if kwarg in kwargs
2037                }
2038                return self._create_component_from_model(
2039                    model=parsed_model, config=config, **(matching_parameters | matching_kwargs)
2040                )
2041            except TypeError as error:
2042                missing_parameters = self._extract_missing_parameters(error)
2043                if missing_parameters:
2044                    raise ValueError(
2045                        f"Error creating component '{type_name}' with parent custom component {model.class_name}: Please provide "
2046                        + ", ".join(
2047                            (
2048                                f"{type_name}.$parameters.{parameter}"
2049                                for parameter in missing_parameters
2050                            )
2051                        )
2052                    )
2053                raise TypeError(
2054                    f"Error creating component '{type_name}' with parent custom component {model.class_name}: {error}"
2055                )
2056        else:
2057            raise ValueError(
2058                f"Error creating custom component {model.class_name}. Subcomponent creation has not been implemented for '{type_name}'"
2059            )
2060
2061    @staticmethod
2062    def _is_component(model_value: Any) -> bool:
2063        return isinstance(model_value, dict) and model_value.get("type") is not None
2064
2065    def create_default_stream(
2066        self, model: DeclarativeStreamModel, config: Config, is_parent: bool = False, **kwargs: Any
2067    ) -> AbstractStream:
2068        primary_key = model.primary_key.__root__ if model.primary_key else None
2069        self._migrate_state(model, config)
2070        self._warn_on_ineffective_incremental_dependency(model)
2071
2072        partition_router = self._build_stream_slicer_from_partition_router(
2073            model.retriever,
2074            config,
2075            stream_name=model.name,
2076            **kwargs,
2077        )
2078        concurrent_cursor = self._build_concurrent_cursor(model, partition_router, config)
2079        if model.incremental_sync and isinstance(model.incremental_sync, DatetimeBasedCursorModel):
2080            cursor_model: DatetimeBasedCursorModel = model.incremental_sync
2081
2082            end_time_option = (
2083                self._create_component_from_model(
2084                    cursor_model.end_time_option, config, parameters=cursor_model.parameters or {}
2085                )
2086                if cursor_model.end_time_option
2087                else None
2088            )
2089            start_time_option = (
2090                self._create_component_from_model(
2091                    cursor_model.start_time_option, config, parameters=cursor_model.parameters or {}
2092                )
2093                if cursor_model.start_time_option
2094                else None
2095            )
2096
2097            datetime_request_options_provider = DatetimeBasedRequestOptionsProvider(
2098                start_time_option=start_time_option,
2099                end_time_option=end_time_option,
2100                partition_field_start=cursor_model.partition_field_start,
2101                partition_field_end=cursor_model.partition_field_end,
2102                config=config,
2103                parameters=model.parameters or {},
2104            )
2105            request_options_provider = (
2106                datetime_request_options_provider
2107                if not isinstance(concurrent_cursor, ConcurrentPerPartitionCursor)
2108                else PerPartitionRequestOptionsProvider(
2109                    partition_router, datetime_request_options_provider
2110                )
2111            )
2112        elif model.incremental_sync and isinstance(
2113            model.incremental_sync, IncrementingCountCursorModel
2114        ):
2115            if isinstance(concurrent_cursor, ConcurrentPerPartitionCursor):
2116                raise ValueError(
2117                    "PerPartition does not support per partition states because switching to global state is time based"
2118                )
2119
2120            cursor_model: IncrementingCountCursorModel = model.incremental_sync  # type: ignore
2121
2122            start_time_option = (
2123                self._create_component_from_model(
2124                    cursor_model.start_value_option,  # type: ignore # mypy still thinks cursor_model of type DatetimeBasedCursor
2125                    config,
2126                    parameters=cursor_model.parameters or {},
2127                )
2128                if cursor_model.start_value_option  # type: ignore # mypy still thinks cursor_model of type DatetimeBasedCursor
2129                else None
2130            )
2131
2132            # The concurrent engine defaults the start/end fields on the slice to "start" and "end", but
2133            # the default DatetimeBasedRequestOptionsProvider() sets them to start_time/end_time
2134            partition_field_start = "start"
2135
2136            request_options_provider = DatetimeBasedRequestOptionsProvider(
2137                start_time_option=start_time_option,
2138                partition_field_start=partition_field_start,
2139                config=config,
2140                parameters=model.parameters or {},
2141            )
2142        else:
2143            request_options_provider = None
2144
2145        transformations = []
2146        if model.transformations:
2147            for transformation_model in model.transformations:
2148                transformations.append(
2149                    self._create_component_from_model(model=transformation_model, config=config)
2150                )
2151        file_uploader = None
2152        if model.file_uploader:
2153            file_uploader = self._create_component_from_model(
2154                model=model.file_uploader, config=config
2155            )
2156
2157        stream_slicer: ConcurrentStreamSlicer = (
2158            partition_router
2159            if isinstance(concurrent_cursor, FinalStateCursor)
2160            else concurrent_cursor
2161        )
2162
2163        retriever = self._create_component_from_model(
2164            model=model.retriever,
2165            config=config,
2166            name=model.name,
2167            primary_key=primary_key,
2168            request_options_provider=request_options_provider,
2169            stream_slicer=stream_slicer,
2170            partition_router=partition_router,
2171            has_stop_condition_cursor=self._is_stop_condition_on_cursor(model),
2172            is_client_side_incremental_sync=self._is_client_side_filtering_enabled(model),
2173            cursor=concurrent_cursor,
2174            transformations=transformations,
2175            file_uploader=file_uploader,
2176            incremental_sync=model.incremental_sync,
2177        )
2178        if isinstance(retriever, AsyncRetriever):
2179            stream_slicer = retriever.stream_slicer
2180
2181        schema_loader: SchemaLoader
2182        if model.schema_loader and isinstance(model.schema_loader, list):
2183            nested_schema_loaders = [
2184                self._create_component_from_model(model=nested_schema_loader, config=config)
2185                for nested_schema_loader in model.schema_loader
2186            ]
2187            schema_loader = CompositeSchemaLoader(
2188                schema_loaders=nested_schema_loaders, parameters={}
2189            )
2190        elif model.schema_loader:
2191            schema_loader = self._create_component_from_model(
2192                model=model.schema_loader,  # type: ignore # If defined, schema_loader is guaranteed not to be a list and will be one of the existing base models
2193                config=config,
2194            )
2195        else:
2196            options = model.parameters or {}
2197            if "name" not in options:
2198                options["name"] = model.name
2199            schema_loader = DefaultSchemaLoader(config=config, parameters=options)
2200        schema_loader = CachingSchemaLoaderDecorator(schema_loader)
2201
2202        stream_name = model.name or ""
2203        return DefaultStream(
2204            partition_generator=StreamSlicerPartitionGenerator(
2205                DeclarativePartitionFactory(
2206                    stream_name,
2207                    schema_loader,
2208                    retriever,
2209                    self._message_repository,
2210                ),
2211                stream_slicer,
2212                slice_limit=self._limit_slices_fetched,
2213            ),
2214            name=stream_name,
2215            json_schema=schema_loader.get_json_schema,
2216            primary_key=get_primary_key_from_stream(primary_key),
2217            cursor_field=(
2218                concurrent_cursor.cursor_field
2219                if hasattr(concurrent_cursor, "cursor_field")
2220                else None
2221            ),
2222            logger=logging.getLogger(f"airbyte.{stream_name}"),
2223            cursor=concurrent_cursor,
2224            supports_file_transfer=hasattr(model, "file_uploader") and bool(model.file_uploader),
2225        )
2226
2227    def _warn_on_ineffective_incremental_dependency(self, model: DeclarativeStreamModel) -> None:
2228        """
2229        `incremental_dependency: true` only takes effect when the substream defines its own
2230        `incremental_sync`: the parent cursor is persisted under the `parent_state` key of the
2231        substream's state, which is only emitted by incremental substreams. On a stream without
2232        `incremental_sync`, the setting is silently ignored and all parent records are re-read on
2233        every sync, so we warn about the misconfiguration instead.
2234        """
2235        if model.incremental_sync:
2236            return
2237
2238        partition_router = getattr(model.retriever, "partition_router", None)
2239        if not partition_router:
2240            return
2241
2242        routers = partition_router if isinstance(partition_router, list) else [partition_router]
2243        for router in routers:
2244            if isinstance(router, GroupingPartitionRouterModel):
2245                router = router.underlying_partition_router
2246            if isinstance(router, SubstreamPartitionRouterModel) and any(
2247                parent_stream_config.incremental_dependency
2248                for parent_stream_config in router.parent_stream_configs
2249            ):
2250                LOGGER.warning(
2251                    f"Stream `{model.name}` has `incremental_dependency: true` in its parent stream configuration but does not define `incremental_sync`. "
2252                    "The parent stream's cursor is only persisted in the state of an incremental substream, so this setting has no effect and all parent records will be re-read on every sync. "
2253                    "Define `incremental_sync` on this stream or remove `incremental_dependency`."
2254                )
2255                return
2256
2257    def _migrate_state(self, model: DeclarativeStreamModel, config: Config) -> None:
2258        stream_name = model.name or ""
2259        stream_state = self._connector_state_manager.get_stream_state(
2260            stream_name=stream_name, namespace=None
2261        )
2262        if model.state_migrations:
2263            state_transformations = [
2264                self._create_component_from_model(state_migration, config, declarative_stream=model)
2265                for state_migration in model.state_migrations
2266            ]
2267        else:
2268            state_transformations = []
2269        stream_state = self.apply_stream_state_migrations(state_transformations, stream_state)
2270        self._connector_state_manager.update_state_for_stream(
2271            stream_name=stream_name, namespace=None, value=stream_state
2272        )
2273
2274    def _is_stop_condition_on_cursor(self, model: DeclarativeStreamModel) -> bool:
2275        return bool(
2276            model.incremental_sync
2277            and hasattr(model.incremental_sync, "is_data_feed")
2278            and model.incremental_sync.is_data_feed
2279        )
2280
2281    def _is_client_side_filtering_enabled(self, model: DeclarativeStreamModel) -> bool:
2282        return bool(
2283            model.incremental_sync
2284            and hasattr(model.incremental_sync, "is_client_side_incremental")
2285            and model.incremental_sync.is_client_side_incremental
2286        )
2287
2288    def _build_stream_slicer_from_partition_router(
2289        self,
2290        model: Union[
2291            AsyncRetrieverModel,
2292            CustomRetrieverModel,
2293            SimpleRetrieverModel,
2294        ],
2295        config: Config,
2296        stream_name: Optional[str] = None,
2297        **kwargs: Any,
2298    ) -> PartitionRouter:
2299        if (
2300            hasattr(model, "partition_router")
2301            and isinstance(model, (SimpleRetrieverModel, AsyncRetrieverModel, CustomRetrieverModel))
2302            and model.partition_router
2303        ):
2304            stream_slicer_model = model.partition_router
2305            if isinstance(stream_slicer_model, list):
2306                return CartesianProductStreamSlicer(
2307                    [
2308                        self._create_component_from_model(
2309                            model=slicer, config=config, stream_name=stream_name or ""
2310                        )
2311                        for slicer in stream_slicer_model
2312                    ],
2313                    parameters={},
2314                )
2315            elif isinstance(stream_slicer_model, dict):
2316                # partition router comes from CustomRetrieverModel therefore has not been parsed as a model
2317                params = stream_slicer_model.get("$parameters")
2318                if not isinstance(params, dict):
2319                    params = {}
2320                    stream_slicer_model["$parameters"] = params
2321
2322                if stream_name is not None:
2323                    params["stream_name"] = stream_name
2324
2325                return self._create_nested_component(  # type: ignore[no-any-return] # There is no guarantee that this will return a stream slicer. If not, we expect an AttributeError during the call to `stream_slices`
2326                    model,
2327                    "partition_router",
2328                    stream_slicer_model,
2329                    config,
2330                    **kwargs,
2331                )
2332            else:
2333                return self._create_component_from_model(  # type: ignore[no-any-return] # Will be created PartitionRouter as stream_slicer_model is model.partition_router
2334                    model=stream_slicer_model, config=config, stream_name=stream_name or ""
2335                )
2336        return SinglePartitionRouter(parameters={})
2337
2338    def _build_concurrent_cursor(
2339        self,
2340        model: DeclarativeStreamModel,
2341        stream_slicer: Optional[PartitionRouter],
2342        config: Config,
2343    ) -> Cursor:
2344        stream_name = model.name or ""
2345        stream_state = self._connector_state_manager.get_stream_state(stream_name, None)
2346
2347        if (
2348            model.incremental_sync
2349            and stream_slicer
2350            and not isinstance(stream_slicer, SinglePartitionRouter)
2351        ):
2352            if isinstance(model.incremental_sync, IncrementingCountCursorModel):
2353                # We don't currently support usage of partition routing and IncrementingCountCursor at the
2354                # same time because we didn't solve for design questions like what the lookback window would
2355                # be as well as global cursor fall backs. We have not seen customers that have needed both
2356                # at the same time yet and are currently punting on this until we need to solve it.
2357                raise ValueError(
2358                    f"The low-code framework does not currently support usage of a PartitionRouter and an IncrementingCountCursor at the same time. Please specify only one of these options for stream {stream_name}."
2359                )
2360            return self.create_concurrent_cursor_from_perpartition_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
2361                state_manager=self._connector_state_manager,
2362                model_type=DatetimeBasedCursorModel,
2363                component_definition=model.incremental_sync.__dict__,
2364                stream_name=stream_name,
2365                stream_state=stream_state,
2366                stream_namespace=None,
2367                config=config or {},
2368                partition_router=stream_slicer,
2369                attempt_to_create_cursor_if_not_provided=True,  # FIXME can we remove that now?
2370            )
2371        elif model.incremental_sync:
2372            if type(model.incremental_sync) == IncrementingCountCursorModel:
2373                return self.create_concurrent_cursor_from_incrementing_count_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
2374                    model_type=IncrementingCountCursorModel,
2375                    component_definition=model.incremental_sync.__dict__,
2376                    stream_name=stream_name,
2377                    stream_namespace=None,
2378                    stream_state=stream_state,
2379                    config=config or {},
2380                )
2381            elif type(model.incremental_sync) == DatetimeBasedCursorModel:
2382                return self.create_concurrent_cursor_from_datetime_based_cursor(  # type: ignore # This is a known issue that we are creating and returning a ConcurrentCursor which does not technically implement the (low-code) StreamSlicer. However, (low-code) StreamSlicer and ConcurrentCursor both implement StreamSlicer.stream_slices() which is the primary method needed for checkpointing
2383                    model_type=type(model.incremental_sync),
2384                    component_definition=model.incremental_sync.__dict__,
2385                    stream_name=stream_name,
2386                    stream_namespace=None,
2387                    stream_state=stream_state,
2388                    config=config or {},
2389                    attempt_to_create_cursor_if_not_provided=True,
2390                )
2391            else:
2392                raise ValueError(
2393                    f"Incremental sync of type {type(model.incremental_sync)} is not supported"
2394                )
2395        return FinalStateCursor(stream_name, None, self._message_repository)
2396
2397    def create_default_error_handler(
2398        self, model: DefaultErrorHandlerModel, config: Config, **kwargs: Any
2399    ) -> DefaultErrorHandler:
2400        backoff_strategies = []
2401        if model.backoff_strategies:
2402            for backoff_strategy_model in model.backoff_strategies:
2403                backoff_strategies.append(
2404                    self._create_component_from_model(model=backoff_strategy_model, config=config)
2405                )
2406
2407        response_filters = []
2408        if model.response_filters:
2409            for response_filter_model in model.response_filters:
2410                response_filters.append(
2411                    self._create_component_from_model(model=response_filter_model, config=config)
2412                )
2413        response_filters.append(
2414            HttpResponseFilter(config=config, parameters=model.parameters or {})
2415        )
2416
2417        return DefaultErrorHandler(
2418            backoff_strategies=backoff_strategies,
2419            max_retries=model.max_retries,
2420            response_filters=response_filters,
2421            config=config,
2422            parameters=model.parameters or {},
2423        )
2424
2425    def create_default_paginator(
2426        self,
2427        model: DefaultPaginatorModel,
2428        config: Config,
2429        *,
2430        url_base: str,
2431        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
2432        decoder: Optional[Decoder] = None,
2433        cursor_used_for_stop_condition: Optional[Cursor] = None,
2434    ) -> Union[DefaultPaginator, PaginatorTestReadDecorator]:
2435        if decoder:
2436            if self._is_supported_decoder_for_pagination(decoder):
2437                decoder_to_use = PaginationDecoderDecorator(decoder=decoder)
2438            else:
2439                raise ValueError(self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(decoder)))
2440        else:
2441            decoder_to_use = PaginationDecoderDecorator(decoder=JsonDecoder(parameters={}))
2442        page_size_option = (
2443            self._create_component_from_model(model=model.page_size_option, config=config)
2444            if model.page_size_option
2445            else None
2446        )
2447        page_token_option = (
2448            self._create_component_from_model(model=model.page_token_option, config=config)
2449            if model.page_token_option
2450            else None
2451        )
2452        pagination_strategy = self._create_component_from_model(
2453            model=model.pagination_strategy,
2454            config=config,
2455            decoder=decoder_to_use,
2456            extractor_model=extractor_model,
2457        )
2458        if cursor_used_for_stop_condition:
2459            pagination_strategy = StopConditionPaginationStrategyDecorator(
2460                pagination_strategy, CursorStopCondition(cursor_used_for_stop_condition)
2461            )
2462        paginator = DefaultPaginator(
2463            decoder=decoder_to_use,
2464            page_size_option=page_size_option,
2465            page_token_option=page_token_option,
2466            pagination_strategy=pagination_strategy,
2467            url_base=url_base,
2468            config=config,
2469            parameters=model.parameters or {},
2470        )
2471        if self._limit_pages_fetched_per_slice:
2472            return PaginatorTestReadDecorator(paginator, self._limit_pages_fetched_per_slice)
2473        return paginator
2474
2475    def create_dpath_extractor(
2476        self,
2477        model: DpathExtractorModel,
2478        config: Config,
2479        decoder: Optional[Decoder] = None,
2480        **kwargs: Any,
2481    ) -> DpathExtractor:
2482        if decoder:
2483            decoder_to_use = decoder
2484        else:
2485            decoder_to_use = JsonDecoder(parameters={})
2486        model_field_path: List[Union[InterpolatedString, str]] = [x for x in model.field_path]
2487
2488        record_expander = None
2489        if model.record_expander:
2490            record_expander = self._create_component_from_model(
2491                model=model.record_expander,
2492                config=config,
2493            )
2494
2495        return DpathExtractor(
2496            decoder=decoder_to_use,
2497            field_path=model_field_path,
2498            config=config,
2499            parameters=model.parameters or {},
2500            record_expander=record_expander,
2501        )
2502
2503    def create_record_expander(
2504        self,
2505        model: RecordExpanderModel,
2506        config: Config,
2507        **kwargs: Any,
2508    ) -> RecordExpander:
2509        truncated_list_retriever = None
2510        suppress_incomplete_fetch_warning = False
2511        if model.truncated_list_retriever:
2512            retriever_model = model.truncated_list_retriever
2513            name = "record_expander_truncated_list"
2514            # `CustomRetriever` allows extra fields, so read from the dumped model to cover both types.
2515            retriever_fields = retriever_model.dict()
2516            for unsupported_option in ("partition_router", "pagination_reset"):
2517                if retriever_fields.get(unsupported_option):
2518                    raise ValueError(
2519                        f"`{unsupported_option}` is not supported on `truncated_list_retriever`."
2520                    )
2521            log_formatter = lambda response: format_http_message(
2522                response,
2523                f"Record expander '{name}' request",
2524                "Request performed in order to fetch the complete nested list of a truncated record.",
2525                name,
2526                is_auxiliary=True,
2527            )
2528            # `name`/`primary_key`/`transformations` are also what a `CustomRetriever` forwards to its
2529            # nested `requester`/`record_selector`, so they are passed for both retriever types.
2530            truncated_list_retriever = self._create_component_from_model(
2531                model=retriever_model,
2532                config=config,
2533                name=name,
2534                primary_key=None,
2535                transformations=[],
2536                log_formatter=log_formatter,
2537            )
2538            # Only a capped paginator makes a shortfall expected; `NoPagination` and retrievers
2539            # without a paginator are never capped.
2540            suppress_incomplete_fetch_warning = isinstance(
2541                truncated_list_retriever, SimpleRetriever
2542            ) and isinstance(truncated_list_retriever.paginator, PaginatorTestReadDecorator)
2543        return RecordExpander(
2544            expand_records_from_field=model.expand_records_from_field,
2545            config=config,
2546            parameters=model.parameters or {},
2547            remain_original_record=model.remain_original_record or False,
2548            on_no_records=OnNoRecords(model.on_no_records.value)
2549            if model.on_no_records
2550            else OnNoRecords.skip,
2551            truncation_indicator_path=model.truncation_indicator_path,
2552            truncated_list_retriever=truncated_list_retriever,
2553            message_repository=self._message_repository,
2554            suppress_incomplete_fetch_warning=suppress_incomplete_fetch_warning,
2555        )
2556
2557    @staticmethod
2558    def create_response_to_file_extractor(
2559        model: ResponseToFileExtractorModel,
2560        **kwargs: Any,
2561    ) -> ResponseToFileExtractor:
2562        return ResponseToFileExtractor(
2563            parameters=model.parameters or {},
2564            preserve_na_values=model.preserve_na_values or False,
2565        )
2566
2567    @staticmethod
2568    def create_exponential_backoff_strategy(
2569        model: ExponentialBackoffStrategyModel, config: Config
2570    ) -> ExponentialBackoffStrategy:
2571        ModelToComponentFactory._validate_jitter_range(model.jitter_range_in_seconds)
2572        return ExponentialBackoffStrategy(
2573            factor=model.factor or 5,
2574            jitter_range_in_seconds=model.jitter_range_in_seconds,
2575            parameters=model.parameters or {},
2576            config=config,
2577        )
2578
2579    @staticmethod
2580    def create_group_by_key(model: GroupByKeyMergeStrategyModel, config: Config) -> GroupByKey:
2581        return GroupByKey(model.key, config=config, parameters=model.parameters or {})
2582
2583    def create_http_requester(
2584        self,
2585        model: HttpRequesterModel,
2586        config: Config,
2587        decoder: Decoder = JsonDecoder(parameters={}),
2588        query_properties_key: Optional[str] = None,
2589        use_cache: Optional[bool] = None,
2590        *,
2591        name: str,
2592    ) -> HttpRequester:
2593        authenticator = (
2594            self._create_component_from_model(
2595                model=model.authenticator,
2596                config=config,
2597                url_base=model.url or model.url_base,
2598                name=name,
2599                decoder=decoder,
2600            )
2601            if model.authenticator
2602            else None
2603        )
2604        error_handler = (
2605            self._create_component_from_model(model=model.error_handler, config=config)
2606            if model.error_handler
2607            else DefaultErrorHandler(
2608                backoff_strategies=[],
2609                response_filters=[],
2610                config=config,
2611                parameters=model.parameters or {},
2612            )
2613        )
2614
2615        api_budget = self._api_budget
2616
2617        request_options_provider = InterpolatedRequestOptionsProvider(
2618            request_body=model.request_body,
2619            request_body_data=model.request_body_data,
2620            request_body_json=model.request_body_json,
2621            request_headers=model.request_headers,
2622            request_parameters=model.request_parameters,  # type: ignore  # QueryProperties have been removed in `create_simple_retriever`
2623            query_properties_key=query_properties_key,
2624            config=config,
2625            parameters=model.parameters or {},
2626        )
2627
2628        assert model.use_cache is not None  # for mypy
2629        assert model.http_method is not None  # for mypy
2630
2631        should_use_cache = (model.use_cache or bool(use_cache)) and not self._disable_cache
2632
2633        return HttpRequester(
2634            name=name,
2635            url=model.url,
2636            url_base=model.url_base,
2637            path=model.path,
2638            authenticator=authenticator,
2639            error_handler=error_handler,
2640            api_budget=api_budget,
2641            http_method=HttpMethod[model.http_method.value],
2642            request_options_provider=request_options_provider,
2643            config=config,
2644            disable_retries=self._disable_retries,
2645            parameters=model.parameters or {},
2646            message_repository=self._message_repository,
2647            use_cache=should_use_cache,
2648            decoder=decoder,
2649            stream_response=decoder.is_stream_response() if decoder else False,
2650        )
2651
2652    @staticmethod
2653    def create_http_response_filter(
2654        model: HttpResponseFilterModel, config: Config, **kwargs: Any
2655    ) -> HttpResponseFilter:
2656        if model.action:
2657            action = ResponseAction(model.action.value)
2658        else:
2659            action = None
2660
2661        failure_type = FailureType(model.failure_type.value) if model.failure_type else None
2662
2663        http_codes = (
2664            set(model.http_codes) if model.http_codes else set()
2665        )  # JSON schema notation has no set data type. The schema enforces an array of unique elements
2666
2667        return HttpResponseFilter(
2668            action=action,
2669            failure_type=failure_type,
2670            error_message=model.error_message or "",
2671            error_message_contains=model.error_message_contains or "",
2672            http_codes=http_codes,
2673            predicate=model.predicate or "",
2674            config=config,
2675            parameters=model.parameters or {},
2676        )
2677
2678    @staticmethod
2679    def create_inline_schema_loader(
2680        model: InlineSchemaLoaderModel, config: Config, **kwargs: Any
2681    ) -> InlineSchemaLoader:
2682        return InlineSchemaLoader(schema=model.schema_ or {}, parameters={})
2683
2684    def create_complex_field_type(
2685        self, model: ComplexFieldTypeModel, config: Config, **kwargs: Any
2686    ) -> ComplexFieldType:
2687        items = (
2688            self._create_component_from_model(model=model.items, config=config)
2689            if isinstance(model.items, ComplexFieldTypeModel)
2690            else model.items
2691        )
2692
2693        return ComplexFieldType(field_type=model.field_type, items=items)
2694
2695    def create_types_map(self, model: TypesMapModel, config: Config, **kwargs: Any) -> TypesMap:
2696        target_type = (
2697            self._create_component_from_model(model=model.target_type, config=config)
2698            if isinstance(model.target_type, ComplexFieldTypeModel)
2699            else model.target_type
2700        )
2701
2702        return TypesMap(
2703            target_type=target_type,
2704            current_type=model.current_type,
2705            condition=model.condition if model.condition is not None else "True",
2706        )
2707
2708    def create_schema_type_identifier(
2709        self, model: SchemaTypeIdentifierModel, config: Config, **kwargs: Any
2710    ) -> SchemaTypeIdentifier:
2711        types_mapping = []
2712        if model.types_mapping:
2713            types_mapping.extend(
2714                [
2715                    self._create_component_from_model(types_map, config=config)
2716                    for types_map in model.types_mapping
2717                ]
2718            )
2719        model_schema_pointer: List[Union[InterpolatedString, str]] = (
2720            [x for x in model.schema_pointer] if model.schema_pointer else []
2721        )
2722        model_key_pointer: List[Union[InterpolatedString, str]] = [x for x in model.key_pointer]
2723        model_type_pointer: Optional[List[Union[InterpolatedString, str]]] = (
2724            [x for x in model.type_pointer] if model.type_pointer else None
2725        )
2726
2727        return SchemaTypeIdentifier(
2728            schema_pointer=model_schema_pointer,
2729            key_pointer=model_key_pointer,
2730            type_pointer=model_type_pointer,
2731            types_mapping=types_mapping,
2732            parameters=model.parameters or {},
2733        )
2734
2735    def create_dynamic_schema_loader(
2736        self, model: DynamicSchemaLoaderModel, config: Config, **kwargs: Any
2737    ) -> DynamicSchemaLoader:
2738        schema_transformations = []
2739        if model.schema_transformations:
2740            for transformation_model in model.schema_transformations:
2741                schema_transformations.append(
2742                    self._create_component_from_model(model=transformation_model, config=config)
2743                )
2744        name = "dynamic_properties"
2745        retriever = self._create_component_from_model(
2746            model=model.retriever,
2747            config=config,
2748            name=name,
2749            primary_key=None,
2750            partition_router=self._build_stream_slicer_from_partition_router(
2751                model.retriever, config
2752            ),
2753            transformations=[],
2754            use_cache=True,
2755            log_formatter=(
2756                lambda response: format_http_message(
2757                    response,
2758                    f"Schema loader '{name}' request",
2759                    f"Request performed in order to extract schema.",
2760                    name,
2761                    is_auxiliary=True,
2762                )
2763            ),
2764        )
2765        schema_type_identifier = self._create_component_from_model(
2766            model.schema_type_identifier, config=config, parameters=model.parameters or {}
2767        )
2768        schema_filter = (
2769            self._create_component_from_model(
2770                model.schema_filter, config=config, parameters=model.parameters or {}
2771            )
2772            if model.schema_filter is not None
2773            else None
2774        )
2775
2776        return DynamicSchemaLoader(
2777            retriever=retriever,
2778            config=config,
2779            schema_transformations=schema_transformations,
2780            schema_filter=schema_filter,
2781            schema_type_identifier=schema_type_identifier,
2782            parameters=model.parameters or {},
2783        )
2784
2785    @staticmethod
2786    def create_json_decoder(model: JsonDecoderModel, config: Config, **kwargs: Any) -> Decoder:
2787        return JsonDecoder(parameters={})
2788
2789    def create_csv_decoder(self, model: CsvDecoderModel, config: Config, **kwargs: Any) -> Decoder:
2790        return CompositeRawDecoder(
2791            parser=ModelToComponentFactory._get_parser(model, config),
2792            stream_response=False if self._emit_connector_builder_messages else True,
2793        )
2794
2795    def create_jsonl_decoder(
2796        self, model: JsonlDecoderModel, config: Config, **kwargs: Any
2797    ) -> Decoder:
2798        return CompositeRawDecoder(
2799            parser=ModelToComponentFactory._get_parser(model, config),
2800            stream_response=False if self._emit_connector_builder_messages else True,
2801        )
2802
2803    def create_json_items_decoder(
2804        self, model: JsonItemsDecoderModel, config: Config, **kwargs: Any
2805    ) -> Decoder:
2806        return CompositeRawDecoder(
2807            parser=ModelToComponentFactory._get_parser(model, config),
2808            stream_response=False if self._emit_connector_builder_messages else True,
2809        )
2810
2811    def create_gzip_decoder(
2812        self, model: GzipDecoderModel, config: Config, **kwargs: Any
2813    ) -> Decoder:
2814        _compressed_response_types = {
2815            "gzip",
2816            "x-gzip",
2817            "gzip, deflate",
2818            "x-gzip, deflate",
2819            "application/zip",
2820            "application/gzip",
2821            "application/x-gzip",
2822            "application/x-zip-compressed",
2823        }
2824
2825        gzip_parser: GzipParser = ModelToComponentFactory._get_parser(model, config)  # type: ignore  # based on the model, we know this will be a GzipParser
2826
2827        if self._emit_connector_builder_messages:
2828            return CompositeRawDecoder(gzip_parser, False)
2829
2830        transport_gzip_parser = GzipParser(inner_parser=gzip_parser)
2831        return CompositeRawDecoder.by_headers(
2832            [
2833                ({"Content-Encoding"}, {"gzip"}, transport_gzip_parser),
2834                ({"Content-Type"}, _compressed_response_types, gzip_parser),
2835            ],
2836            stream_response=True,
2837            fallback_parser=gzip_parser,
2838        )
2839
2840    @staticmethod
2841    def create_iterable_decoder(
2842        model: IterableDecoderModel, config: Config, **kwargs: Any
2843    ) -> IterableDecoder:
2844        return IterableDecoder(parameters={})
2845
2846    @staticmethod
2847    def create_xml_decoder(model: XmlDecoderModel, config: Config, **kwargs: Any) -> XmlDecoder:
2848        return XmlDecoder(parameters={})
2849
2850    def create_zipfile_decoder(
2851        self, model: ZipfileDecoderModel, config: Config, **kwargs: Any
2852    ) -> ZipfileDecoder:
2853        return ZipfileDecoder(parser=ModelToComponentFactory._get_parser(model.decoder, config))
2854
2855    @staticmethod
2856    def _get_parser(model: BaseModel, config: Config) -> Parser:
2857        if isinstance(model, JsonDecoderModel):
2858            # Note that the logic is a bit different from the JsonDecoder as there is some legacy that is maintained to return {} on error cases
2859            return JsonParser()
2860        elif isinstance(model, JsonItemsDecoderModel):
2861            return JsonItemsParser(
2862                items_path=model.items_path,
2863                encoding=model.encoding,
2864            )
2865        elif isinstance(model, JsonlDecoderModel):
2866            return JsonLineParser()
2867        elif isinstance(model, CsvDecoderModel):
2868            return CsvParser(
2869                encoding=model.encoding,
2870                delimiter=model.delimiter,
2871                set_values_to_none=model.set_values_to_none,
2872            )
2873        elif isinstance(model, GzipDecoderModel):
2874            return GzipParser(
2875                inner_parser=ModelToComponentFactory._get_parser(model.decoder, config)
2876            )
2877        elif isinstance(
2878            model, (CustomDecoderModel, IterableDecoderModel, XmlDecoderModel, ZipfileDecoderModel)
2879        ):
2880            raise ValueError(f"Decoder type {model} does not have parser associated to it")
2881
2882        raise ValueError(f"Unknown decoder type {model}")
2883
2884    @staticmethod
2885    def create_json_file_schema_loader(
2886        model: JsonFileSchemaLoaderModel, config: Config, **kwargs: Any
2887    ) -> JsonFileSchemaLoader:
2888        return JsonFileSchemaLoader(
2889            file_path=model.file_path or "", config=config, parameters=model.parameters or {}
2890        )
2891
2892    def create_jwt_authenticator(
2893        self, model: JwtAuthenticatorModel, config: Config, **kwargs: Any
2894    ) -> JwtAuthenticator:
2895        jwt_headers = model.jwt_headers or JwtHeadersModel(kid=None, typ="JWT", cty=None)
2896        jwt_payload = model.jwt_payload or JwtPayloadModel(iss=None, sub=None, aud=None)
2897        request_option = (
2898            self._create_component_from_model(model.request_option, config)
2899            if model.request_option
2900            else None
2901        )
2902        return JwtAuthenticator(
2903            config=config,
2904            parameters=model.parameters or {},
2905            algorithm=JwtAlgorithm(model.algorithm.value),
2906            secret_key=model.secret_key,
2907            base64_encode_secret_key=model.base64_encode_secret_key,
2908            token_duration=model.token_duration,
2909            header_prefix=model.header_prefix,
2910            kid=jwt_headers.kid,
2911            typ=jwt_headers.typ,
2912            cty=jwt_headers.cty,
2913            iss=jwt_payload.iss,
2914            sub=jwt_payload.sub,
2915            aud=jwt_payload.aud,
2916            additional_jwt_headers=model.additional_jwt_headers,
2917            additional_jwt_payload=model.additional_jwt_payload,
2918            passphrase=model.passphrase,
2919            request_option=request_option,
2920        )
2921
2922    def create_list_partition_router(
2923        self, model: ListPartitionRouterModel, config: Config, **kwargs: Any
2924    ) -> ListPartitionRouter:
2925        request_option = (
2926            self._create_component_from_model(model.request_option, config)
2927            if model.request_option
2928            else None
2929        )
2930        return ListPartitionRouter(
2931            cursor_field=model.cursor_field,
2932            request_option=request_option,
2933            values=model.values,
2934            config=config,
2935            parameters=model.parameters or {},
2936        )
2937
2938    @staticmethod
2939    def create_min_max_datetime(
2940        model: MinMaxDatetimeModel, config: Config, **kwargs: Any
2941    ) -> MinMaxDatetime:
2942        return MinMaxDatetime(
2943            datetime=model.datetime,
2944            datetime_format=model.datetime_format or "",
2945            max_datetime=model.max_datetime or "",
2946            min_datetime=model.min_datetime or "",
2947            parameters=model.parameters or {},
2948        )
2949
2950    @staticmethod
2951    def create_no_auth(model: NoAuthModel, config: Config, **kwargs: Any) -> NoAuth:
2952        return NoAuth(parameters=model.parameters or {})
2953
2954    @staticmethod
2955    def create_no_pagination(
2956        model: NoPaginationModel, config: Config, **kwargs: Any
2957    ) -> NoPagination:
2958        return NoPagination(parameters={})
2959
2960    def create_oauth_authenticator(
2961        self, model: OAuthAuthenticatorModel, config: Config, **kwargs: Any
2962    ) -> DeclarativeOauth2Authenticator:
2963        profile_assertion = (
2964            self._create_component_from_model(model.profile_assertion, config=config)
2965            if model.profile_assertion
2966            else None
2967        )
2968
2969        refresh_token_error_status_codes, refresh_token_error_key, refresh_token_error_values = (
2970            self._get_refresh_token_error_information(model)
2971        )
2972        if model.refresh_token_updater:
2973            # ignore type error because fixing it would have a lot of dependencies, revisit later
2974            return DeclarativeSingleUseRefreshTokenOauth2Authenticator(  # type: ignore
2975                config,
2976                InterpolatedString.create(
2977                    model.token_refresh_endpoint,  # type: ignore
2978                    parameters=model.parameters or {},
2979                ).eval(config),
2980                access_token_name=InterpolatedString.create(
2981                    model.access_token_name or "access_token", parameters=model.parameters or {}
2982                ).eval(config),
2983                refresh_token_name=model.refresh_token_updater.refresh_token_name,
2984                expires_in_name=InterpolatedString.create(
2985                    model.expires_in_name or "expires_in", parameters=model.parameters or {}
2986                ).eval(config),
2987                client_id_name=InterpolatedString.create(
2988                    model.client_id_name or "client_id", parameters=model.parameters or {}
2989                ).eval(config),
2990                client_id=InterpolatedString.create(
2991                    model.client_id, parameters=model.parameters or {}
2992                ).eval(config)
2993                if model.client_id
2994                else model.client_id,
2995                client_secret_name=InterpolatedString.create(
2996                    model.client_secret_name or "client_secret", parameters=model.parameters or {}
2997                ).eval(config),
2998                client_secret=InterpolatedString.create(
2999                    model.client_secret, parameters=model.parameters or {}
3000                ).eval(config)
3001                if model.client_secret
3002                else model.client_secret,
3003                access_token_config_path=model.refresh_token_updater.access_token_config_path,
3004                refresh_token_config_path=model.refresh_token_updater.refresh_token_config_path,
3005                token_expiry_date_config_path=model.refresh_token_updater.token_expiry_date_config_path,
3006                grant_type_name=InterpolatedString.create(
3007                    model.grant_type_name or "grant_type", parameters=model.parameters or {}
3008                ).eval(config),
3009                grant_type=InterpolatedString.create(
3010                    model.grant_type or "refresh_token", parameters=model.parameters or {}
3011                ).eval(config),
3012                refresh_request_body=InterpolatedMapping(
3013                    model.refresh_request_body or {}, parameters=model.parameters or {}
3014                ).eval(config),
3015                refresh_request_headers=InterpolatedMapping(
3016                    model.refresh_request_headers or {}, parameters=model.parameters or {}
3017                ).eval(config),
3018                send_refresh_request_as_query_params=bool(
3019                    model.send_refresh_request_as_query_params
3020                ),
3021                scopes=model.scopes,
3022                token_expiry_date_format=model.token_expiry_date_format,
3023                token_expiry_is_time_of_expiration=bool(model.token_expiry_date_format),
3024                message_repository=self._message_repository,
3025                refresh_token_error_status_codes=refresh_token_error_status_codes,
3026                refresh_token_error_key=refresh_token_error_key,
3027                refresh_token_error_values=refresh_token_error_values,
3028            )
3029        # ignore type error because fixing it would have a lot of dependencies, revisit later
3030        return DeclarativeOauth2Authenticator(  # type: ignore
3031            access_token_name=model.access_token_name or "access_token",
3032            access_token_value=model.access_token_value,
3033            client_id_name=model.client_id_name or "client_id",
3034            client_id=model.client_id,
3035            client_secret_name=model.client_secret_name or "client_secret",
3036            client_secret=model.client_secret,
3037            expires_in_name=model.expires_in_name or "expires_in",
3038            grant_type_name=model.grant_type_name or "grant_type",
3039            grant_type=model.grant_type or "refresh_token",
3040            refresh_request_body=model.refresh_request_body,
3041            refresh_request_headers=model.refresh_request_headers,
3042            send_refresh_request_as_query_params=bool(model.send_refresh_request_as_query_params),
3043            refresh_token_name=model.refresh_token_name or "refresh_token",
3044            refresh_token=model.refresh_token,
3045            scopes=model.scopes,
3046            token_expiry_date=model.token_expiry_date,
3047            token_expiry_date_format=model.token_expiry_date_format,
3048            token_expiry_is_time_of_expiration=bool(model.token_expiry_date_format),
3049            token_refresh_endpoint=model.token_refresh_endpoint,
3050            config=config,
3051            parameters=model.parameters or {},
3052            message_repository=self._message_repository,
3053            profile_assertion=profile_assertion,
3054            use_profile_assertion=model.use_profile_assertion,
3055            refresh_token_error_status_codes=refresh_token_error_status_codes,
3056            refresh_token_error_key=refresh_token_error_key,
3057            refresh_token_error_values=refresh_token_error_values,
3058        )
3059
3060    @staticmethod
3061    def _get_refresh_token_error_information(
3062        model: OAuthAuthenticatorModel,
3063    ) -> Tuple[Tuple[int, ...], str, Tuple[str, ...]]:
3064        """
3065        In a previous version of the CDK, the auth error as config_error was only done if a refresh token updater was
3066        defined. As a transition, we added those fields on the OAuthAuthenticatorModel. This method ensures that the
3067        information is defined only once and return the right fields.
3068        """
3069        refresh_token_updater = model.refresh_token_updater
3070        is_defined_on_refresh_token_updated = refresh_token_updater and (
3071            refresh_token_updater.refresh_token_error_status_codes
3072            or refresh_token_updater.refresh_token_error_key
3073            or refresh_token_updater.refresh_token_error_values
3074        )
3075        is_defined_on_oauth_authenticator = (
3076            model.refresh_token_error_status_codes
3077            or model.refresh_token_error_key
3078            or model.refresh_token_error_values
3079        )
3080        if is_defined_on_refresh_token_updated and is_defined_on_oauth_authenticator:
3081            raise ValueError(
3082                "refresh_token_error should either be defined on the OAuthAuthenticatorModel or the RefreshTokenUpdaterModel, not both"
3083            )
3084
3085        if is_defined_on_refresh_token_updated:
3086            not_optional_refresh_token_updater: RefreshTokenUpdaterModel = refresh_token_updater  # type: ignore  # we know from the condition that this is not None
3087            return (
3088                tuple(not_optional_refresh_token_updater.refresh_token_error_status_codes)
3089                if not_optional_refresh_token_updater.refresh_token_error_status_codes
3090                else (),
3091                not_optional_refresh_token_updater.refresh_token_error_key or "",
3092                tuple(not_optional_refresh_token_updater.refresh_token_error_values)
3093                if not_optional_refresh_token_updater.refresh_token_error_values
3094                else (),
3095            )
3096        elif is_defined_on_oauth_authenticator:
3097            return (
3098                tuple(model.refresh_token_error_status_codes)
3099                if model.refresh_token_error_status_codes
3100                else (),
3101                model.refresh_token_error_key or "",
3102                tuple(model.refresh_token_error_values) if model.refresh_token_error_values else (),
3103            )
3104
3105        # returning default values we think cover most cases
3106        return (400,), "error", ("invalid_grant", "invalid_permissions")
3107
3108    def create_offset_increment(
3109        self,
3110        model: OffsetIncrementModel,
3111        config: Config,
3112        decoder: Decoder,
3113        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
3114        **kwargs: Any,
3115    ) -> OffsetIncrement:
3116        if isinstance(decoder, PaginationDecoderDecorator):
3117            inner_decoder = decoder.decoder
3118        else:
3119            inner_decoder = decoder
3120            decoder = PaginationDecoderDecorator(decoder=decoder)
3121
3122        if self._is_supported_decoder_for_pagination(inner_decoder):
3123            decoder_to_use = decoder
3124        else:
3125            raise ValueError(
3126                self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(inner_decoder))
3127            )
3128
3129        # Ideally we would instantiate the runtime extractor from highest most level (in this case the SimpleRetriever)
3130        # so that it can be shared by OffSetIncrement and RecordSelector. However, due to how we instantiate the
3131        # decoder with various decorators here, but not in create_record_selector, it is simpler to retain existing
3132        # behavior by having two separate extractors with identical behavior since they use the same extractor model.
3133        # When we have more time to investigate we can look into reusing the same component.
3134        extractor = (
3135            self._create_component_from_model(
3136                model=extractor_model, config=config, decoder=decoder_to_use
3137            )
3138            if extractor_model
3139            else None
3140        )
3141
3142        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
3143        # If page_size is a string that represents an integer (not an interpolation), convert it back.
3144        page_size = model.page_size
3145        if isinstance(page_size, str) and page_size.isdigit():
3146            page_size = int(page_size)
3147
3148        return OffsetIncrement(
3149            page_size=page_size,
3150            config=config,
3151            decoder=decoder_to_use,
3152            extractor=extractor,
3153            inject_on_first_request=model.inject_on_first_request or False,
3154            parameters=model.parameters or {},
3155        )
3156
3157    def create_page_increment(
3158        self,
3159        model: PageIncrementModel,
3160        config: Config,
3161        decoder: Optional[Decoder] = None,
3162        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
3163        **kwargs: Any,
3164    ) -> PageIncrement:
3165        # Like OffsetIncrement, we instantiate a separate extractor with identical behavior to the
3166        # RecordSelector's so the strategy can count the raw records in the response. This ensures
3167        # pagination is driven by the API's page size, not the post-filter record count.
3168        extractor = (
3169            self._create_component_from_model(model=extractor_model, config=config, decoder=decoder)
3170            if extractor_model
3171            else None
3172        )
3173
3174        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
3175        # If page_size is a string that represents an integer (not an interpolation), convert it back.
3176        page_size = model.page_size
3177        if isinstance(page_size, str) and page_size.isdigit():
3178            page_size = int(page_size)
3179
3180        return PageIncrement(
3181            page_size=page_size,
3182            config=config,
3183            start_from_page=model.start_from_page or 0,
3184            inject_on_first_request=model.inject_on_first_request or False,
3185            extractor=extractor,
3186            parameters=model.parameters or {},
3187        )
3188
3189    def create_parent_stream_config(
3190        self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any
3191    ) -> ParentStreamConfig:
3192        declarative_stream = self._create_component_from_model(
3193            model.stream,
3194            config=config,
3195            is_parent=True,
3196            **kwargs,
3197        )
3198        request_option = (
3199            self._create_component_from_model(model.request_option, config=config)
3200            if model.request_option
3201            else None
3202        )
3203
3204        if model.lazy_read_pointer and any("*" in pointer for pointer in model.lazy_read_pointer):
3205            raise ValueError(
3206                "The '*' wildcard in 'lazy_read_pointer' is not supported — only direct paths are allowed."
3207            )
3208
3209        model_lazy_read_pointer: List[Union[InterpolatedString, str]] = (
3210            [x for x in model.lazy_read_pointer] if model.lazy_read_pointer else []
3211        )
3212
3213        return ParentStreamConfig(
3214            parent_key=model.parent_key,
3215            request_option=request_option,
3216            stream=declarative_stream,
3217            partition_field=model.partition_field,
3218            config=config,
3219            incremental_dependency=model.incremental_dependency or False,
3220            parameters=model.parameters or {},
3221            extra_fields=model.extra_fields,
3222            lazy_read_pointer=model_lazy_read_pointer,
3223        )
3224
3225    def create_properties_from_endpoint(
3226        self, model: PropertiesFromEndpointModel, config: Config, **kwargs: Any
3227    ) -> PropertiesFromEndpoint:
3228        retriever = self._create_component_from_model(
3229            model=model.retriever,
3230            config=config,
3231            name="dynamic_properties",
3232            primary_key=None,
3233            stream_slicer=None,
3234            transformations=[],
3235            use_cache=True,  # Enable caching on the HttpRequester/HttpClient because the properties endpoint will be called for every slice being processed, and it is highly unlikely for the response to different
3236        )
3237        return PropertiesFromEndpoint(
3238            property_field_path=model.property_field_path,
3239            retriever=retriever,
3240            config=config,
3241            parameters=model.parameters or {},
3242        )
3243
3244    def create_property_chunking(
3245        self, model: PropertyChunkingModel, config: Config, **kwargs: Any
3246    ) -> PropertyChunking:
3247        record_merge_strategy = (
3248            self._create_component_from_model(
3249                model=model.record_merge_strategy, config=config, **kwargs
3250            )
3251            if model.record_merge_strategy
3252            else None
3253        )
3254
3255        property_limit_type: PropertyLimitType
3256        match model.property_limit_type:
3257            case PropertyLimitTypeModel.property_count:
3258                property_limit_type = PropertyLimitType.property_count
3259            case PropertyLimitTypeModel.characters:
3260                property_limit_type = PropertyLimitType.characters
3261            case _:
3262                raise ValueError(f"Invalid PropertyLimitType {property_limit_type}")
3263
3264        return PropertyChunking(
3265            property_limit_type=property_limit_type,
3266            property_limit=model.property_limit,
3267            record_merge_strategy=record_merge_strategy,
3268            config=config,
3269            parameters=model.parameters or {},
3270        )
3271
3272    def create_query_properties(
3273        self, model: QueryPropertiesModel, config: Config, *, stream_name: str, **kwargs: Any
3274    ) -> QueryProperties:
3275        if isinstance(model.property_list, list):
3276            property_list = model.property_list
3277        else:
3278            property_list = self._create_component_from_model(
3279                model=model.property_list, config=config, **kwargs
3280            )
3281
3282        property_chunking = (
3283            self._create_component_from_model(
3284                model=model.property_chunking, config=config, **kwargs
3285            )
3286            if model.property_chunking
3287            else None
3288        )
3289
3290        property_selector = (
3291            self._create_component_from_model(
3292                model=model.property_selector, config=config, stream_name=stream_name, **kwargs
3293            )
3294            if model.property_selector
3295            else None
3296        )
3297
3298        return QueryProperties(
3299            property_list=property_list,
3300            always_include_properties=model.always_include_properties,
3301            property_chunking=property_chunking,
3302            property_selector=property_selector,
3303            config=config,
3304            parameters=model.parameters or {},
3305        )
3306
3307    def create_json_schema_property_selector(
3308        self,
3309        model: JsonSchemaPropertySelectorModel,
3310        config: Config,
3311        *,
3312        stream_name: str,
3313        **kwargs: Any,
3314    ) -> JsonSchemaPropertySelector:
3315        configured_stream = self._stream_name_to_configured_stream.get(stream_name)
3316
3317        transformations = []
3318        if model.transformations:
3319            for transformation_model in model.transformations:
3320                transformations.append(
3321                    self._create_component_from_model(model=transformation_model, config=config)
3322                )
3323
3324        return JsonSchemaPropertySelector(
3325            configured_stream=configured_stream,
3326            properties_transformations=transformations,
3327            config=config,
3328            parameters=model.parameters or {},
3329        )
3330
3331    @staticmethod
3332    def create_record_filter(
3333        model: RecordFilterModel, config: Config, **kwargs: Any
3334    ) -> RecordFilter:
3335        return RecordFilter(
3336            condition=model.condition or "", config=config, parameters=model.parameters or {}
3337        )
3338
3339    @staticmethod
3340    def create_request_path(model: RequestPathModel, config: Config, **kwargs: Any) -> RequestPath:
3341        return RequestPath(parameters={})
3342
3343    @staticmethod
3344    def create_request_option(
3345        model: RequestOptionModel, config: Config, **kwargs: Any
3346    ) -> RequestOption:
3347        inject_into = RequestOptionType(model.inject_into.value)
3348        field_path: Optional[List[Union[InterpolatedString, str]]] = (
3349            [
3350                InterpolatedString.create(segment, parameters=kwargs.get("parameters", {}))
3351                for segment in model.field_path
3352            ]
3353            if model.field_path
3354            else None
3355        )
3356        field_name = (
3357            InterpolatedString.create(model.field_name, parameters=kwargs.get("parameters", {}))
3358            if model.field_name
3359            else None
3360        )
3361        return RequestOption(
3362            field_name=field_name,
3363            field_path=field_path,
3364            inject_into=inject_into,
3365            parameters=kwargs.get("parameters", {}),
3366        )
3367
3368    def create_record_selector(
3369        self,
3370        model: RecordSelectorModel,
3371        config: Config,
3372        *,
3373        name: str,
3374        transformations: List[RecordTransformation] | None = None,
3375        decoder: Decoder | None = None,
3376        client_side_incremental_sync_cursor: Optional[Cursor] = None,
3377        is_client_side_incremental_sync: bool = False,
3378        file_uploader: Optional[DefaultFileUploader] = None,
3379        **kwargs: Any,
3380    ) -> RecordSelector:
3381        extractor = self._create_component_from_model(
3382            model=model.extractor, decoder=decoder, config=config
3383        )
3384        record_filter = (
3385            self._create_component_from_model(model.record_filter, config=config)
3386            if model.record_filter
3387            else None
3388        )
3389
3390        # A client-side incremental stream transforms before filtering by default. That default belongs to the flag,
3391        # not to the component that ends up doing the cursor comparison: a data feed does it in the retriever and
3392        # receives no cursor here, but its `record_filter` condition must keep running after the transformations.
3393        default_transform_before_filtering = bool(
3394            client_side_incremental_sync_cursor or is_client_side_incremental_sync
3395        )
3396        transform_before_filtering = (
3397            default_transform_before_filtering
3398            if model.transform_before_filtering is None
3399            else model.transform_before_filtering
3400        )
3401        if client_side_incremental_sync_cursor:
3402            record_filter = ClientSideIncrementalRecordFilterDecorator(
3403                config=config,
3404                parameters=model.parameters,
3405                condition=model.record_filter.condition
3406                if (model.record_filter and hasattr(model.record_filter, "condition"))
3407                else None,
3408                cursor=client_side_incremental_sync_cursor,
3409            )
3410
3411        if model.schema_normalization is None:
3412            # default to no schema normalization if not set
3413            model.schema_normalization = SchemaNormalizationModel.None_
3414
3415        schema_normalization = (
3416            TypeTransformer(SCHEMA_TRANSFORMER_TYPE_MAPPING[model.schema_normalization])
3417            if isinstance(model.schema_normalization, SchemaNormalizationModel)
3418            else self._create_component_from_model(model.schema_normalization, config=config)  # type: ignore[arg-type] # custom normalization model expected here
3419        )
3420
3421        return RecordSelector(
3422            extractor=extractor,
3423            name=name,
3424            config=config,
3425            record_filter=record_filter,
3426            transformations=transformations or [],
3427            file_uploader=file_uploader,
3428            schema_normalization=schema_normalization,
3429            parameters=model.parameters or {},
3430            transform_before_filtering=transform_before_filtering,
3431        )
3432
3433    @staticmethod
3434    def create_remove_fields(
3435        model: RemoveFieldsModel, config: Config, **kwargs: Any
3436    ) -> RemoveFields:
3437        return RemoveFields(
3438            field_pointers=model.field_pointers, condition=model.condition or "", parameters={}
3439        )
3440
3441    def create_selective_authenticator(
3442        self, model: SelectiveAuthenticatorModel, config: Config, **kwargs: Any
3443    ) -> DeclarativeAuthenticator:
3444        authenticators = {
3445            name: self._create_component_from_model(model=auth, config=config)
3446            for name, auth in model.authenticators.items()
3447        }
3448        # SelectiveAuthenticator will return instance of DeclarativeAuthenticator or raise ValueError error
3449        return SelectiveAuthenticator(  # type: ignore[abstract]
3450            config=config,
3451            authenticators=authenticators,
3452            authenticator_selection_path=model.authenticator_selection_path,
3453            **kwargs,
3454        )
3455
3456    @staticmethod
3457    def create_legacy_session_token_authenticator(
3458        model: LegacySessionTokenAuthenticatorModel, config: Config, *, url_base: str, **kwargs: Any
3459    ) -> LegacySessionTokenAuthenticator:
3460        return LegacySessionTokenAuthenticator(
3461            api_url=url_base,
3462            header=model.header,
3463            login_url=model.login_url,
3464            password=model.password or "",
3465            session_token=model.session_token or "",
3466            session_token_response_key=model.session_token_response_key or "",
3467            username=model.username or "",
3468            validate_session_url=model.validate_session_url,
3469            config=config,
3470            parameters=model.parameters or {},
3471        )
3472
3473    def create_simple_retriever(
3474        self,
3475        model: SimpleRetrieverModel,
3476        config: Config,
3477        *,
3478        name: str,
3479        primary_key: Optional[Union[str, List[str], List[List[str]]]],
3480        request_options_provider: Optional[RequestOptionsProvider] = None,
3481        cursor: Optional[Cursor] = None,
3482        has_stop_condition_cursor: bool = False,
3483        is_client_side_incremental_sync: bool = False,
3484        transformations: List[RecordTransformation],
3485        file_uploader: Optional[DefaultFileUploader] = None,
3486        incremental_sync: Optional[
3487            Union[IncrementingCountCursorModel, DatetimeBasedCursorModel]
3488        ] = None,
3489        use_cache: Optional[bool] = None,
3490        log_formatter: Optional[Callable[[Response], Any]] = None,
3491        partition_router: Optional[PartitionRouter] = None,
3492        **kwargs: Any,
3493    ) -> SimpleRetriever:
3494        def _get_url(req: Requester) -> str:
3495            """
3496            Closure to get the URL from the requester. This is used to get the URL in the case of a lazy retriever.
3497            This is needed because the URL is not set until the requester is created.
3498            """
3499
3500            _url: str = (
3501                model.requester.url
3502                if hasattr(model.requester, "url") and model.requester.url is not None
3503                else req.get_url(stream_state=None, stream_slice=None, next_page_token=None)
3504            )
3505            _url_base: str = (
3506                model.requester.url_base
3507                if hasattr(model.requester, "url_base") and model.requester.url_base is not None
3508                else req.get_url_base(stream_state=None, stream_slice=None, next_page_token=None)
3509            )
3510
3511            return _url or _url_base
3512
3513        if cursor is None:
3514            cursor = FinalStateCursor(name, None, self._message_repository)
3515
3516        # A data feed drops the records the cursor considers already synced in the retriever, which sits downstream of
3517        # the paginator. Letting the record selector drop them as well would be redundant and would hide them from the
3518        # pagination stop condition, so a data feed never delegates that filtering to the record selector, whether
3519        # `is_client_side_incremental` is set or not. The `condition` from `record_filter` is intentionally left out of
3520        # the post-pagination filter and stays in the record selector, which preserves the existing behaviour: the
3521        # selector runs inside the page loop, so the records the condition rejects never reach the paginator's
3522        # accounting. Moving it downstream would start counting them.
3523        post_pagination_filter = (
3524            ClientSideIncrementalRecordFilterDecorator(
3525                config=config,
3526                parameters=model.parameters or {},
3527                condition=None,
3528                cursor=cursor,
3529            )
3530            if has_stop_condition_cursor
3531            else None
3532        )
3533        client_side_incremental_cursor = (
3534            cursor if is_client_side_incremental_sync and not post_pagination_filter else None
3535        )
3536        if post_pagination_filter and is_client_side_incremental_sync:
3537            LOGGER.warning(
3538                f"Stream {name}: `is_client_side_incremental` adds no record filtering when `is_data_feed` is set, "
3539                "as a data feed already filters on the cursor value. It still makes the record selector apply the "
3540                "transformations before the `record_filter` condition."
3541            )
3542
3543        decoder = (
3544            self._create_component_from_model(model=model.decoder, config=config)
3545            if model.decoder
3546            else JsonDecoder(parameters={})
3547        )
3548        record_selector = self._create_component_from_model(
3549            model=model.record_selector,
3550            name=name,
3551            config=config,
3552            decoder=decoder,
3553            transformations=transformations,
3554            client_side_incremental_sync_cursor=client_side_incremental_cursor,
3555            is_client_side_incremental_sync=is_client_side_incremental_sync,
3556            file_uploader=file_uploader,
3557        )
3558
3559        query_properties: Optional[QueryProperties] = None
3560        query_properties_key: Optional[str] = None
3561        self._ensure_query_properties_to_model(model.requester)
3562        if self._has_query_properties_in_request_parameters(model.requester):
3563            # It is better to be explicit about an error if PropertiesFromEndpoint is defined in multiple
3564            # places instead of default to request_parameters which isn't clearly documented
3565            if (
3566                hasattr(model.requester, "fetch_properties_from_endpoint")
3567                and model.requester.fetch_properties_from_endpoint
3568            ):
3569                raise ValueError(
3570                    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"
3571                )
3572
3573            query_properties_definitions = []
3574            for key, request_parameter in model.requester.request_parameters.items():  # type: ignore # request_parameters is already validated to be a Mapping using _has_query_properties_in_request_parameters()
3575                if isinstance(request_parameter, QueryPropertiesModel):
3576                    query_properties_key = key
3577                    query_properties_definitions.append(request_parameter)
3578
3579            if len(query_properties_definitions) > 1:
3580                raise ValueError(
3581                    f"request_parameters only supports defining one QueryProperties field, but found {len(query_properties_definitions)} usages"
3582                )
3583
3584            if len(query_properties_definitions) == 1:
3585                query_properties = self._create_component_from_model(
3586                    model=query_properties_definitions[0], stream_name=name, config=config
3587                )
3588
3589            # Removes QueryProperties components from the interpolated mappings because it has been designed
3590            # to be used by the SimpleRetriever and will be resolved from the provider from the slice directly
3591            # instead of through jinja interpolation
3592            if hasattr(model.requester, "request_parameters") and isinstance(
3593                model.requester.request_parameters, Mapping
3594            ):
3595                model.requester.request_parameters = self._remove_query_properties(
3596                    model.requester.request_parameters
3597                )
3598        elif (
3599            hasattr(model.requester, "fetch_properties_from_endpoint")
3600            and model.requester.fetch_properties_from_endpoint
3601        ):
3602            # todo: Deprecate this condition once dependent connectors migrate to query_properties
3603            query_properties_definition = QueryPropertiesModel(
3604                type="QueryProperties",
3605                property_list=model.requester.fetch_properties_from_endpoint,
3606                always_include_properties=None,
3607                property_chunking=None,
3608            )  # type: ignore # $parameters has a default value
3609
3610            query_properties = self.create_query_properties(
3611                model=query_properties_definition,
3612                stream_name=name,
3613                config=config,
3614            )
3615        elif hasattr(model.requester, "query_properties") and model.requester.query_properties:
3616            query_properties = self.create_query_properties(
3617                model=model.requester.query_properties,
3618                stream_name=name,
3619                config=config,
3620            )
3621
3622        requester = self._create_component_from_model(
3623            model=model.requester,
3624            decoder=decoder,
3625            name=name,
3626            query_properties_key=query_properties_key,
3627            use_cache=use_cache,
3628            config=config,
3629        )
3630
3631        if not request_options_provider:
3632            request_options_provider = DefaultRequestOptionsProvider(parameters={})
3633        if isinstance(request_options_provider, DefaultRequestOptionsProvider) and isinstance(
3634            partition_router, PartitionRouter
3635        ):
3636            request_options_provider = partition_router
3637
3638        paginator = (
3639            self._create_component_from_model(
3640                model=model.paginator,
3641                config=config,
3642                url_base=_get_url(requester),
3643                extractor_model=model.record_selector.extractor,
3644                decoder=decoder,
3645                cursor_used_for_stop_condition=cursor if has_stop_condition_cursor else None,
3646            )
3647            if model.paginator
3648            else NoPagination(parameters={})
3649        )
3650
3651        ignore_stream_slicer_parameters_on_paginated_requests = (
3652            model.ignore_stream_slicer_parameters_on_paginated_requests or False
3653        )
3654
3655        if (
3656            model.partition_router
3657            and isinstance(model.partition_router, SubstreamPartitionRouterModel)
3658            and not bool(self._connector_state_manager.get_stream_state(name, None))
3659            and any(
3660                parent_stream_config.lazy_read_pointer
3661                for parent_stream_config in model.partition_router.parent_stream_configs
3662            )
3663        ):
3664            if incremental_sync:
3665                if incremental_sync.type != "DatetimeBasedCursor":
3666                    raise ValueError(
3667                        f"LazySimpleRetriever only supports DatetimeBasedCursor. Found: {incremental_sync.type}."
3668                    )
3669
3670                elif incremental_sync.step or incremental_sync.cursor_granularity:
3671                    raise ValueError(
3672                        f"Found more that one slice per parent. LazySimpleRetriever only supports single slice read for stream - {name}."
3673                    )
3674
3675            if model.decoder and model.decoder.type != "JsonDecoder":
3676                raise ValueError(
3677                    f"LazySimpleRetriever only supports JsonDecoder. Found: {model.decoder.type}."
3678                )
3679
3680            return LazySimpleRetriever(
3681                name=name,
3682                paginator=paginator,
3683                primary_key=primary_key,
3684                requester=requester,
3685                record_selector=record_selector,
3686                stream_slicer=_NO_STREAM_SLICING,
3687                request_option_provider=request_options_provider,
3688                config=config,
3689                ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
3690                post_pagination_filter=post_pagination_filter,
3691                parameters=model.parameters or {},
3692            )
3693
3694        if (
3695            model.record_selector.record_filter
3696            and model.pagination_reset
3697            and model.pagination_reset.limits
3698        ):
3699            raise ValueError("PaginationResetLimits are not supported while having record filter.")
3700
3701        return SimpleRetriever(
3702            name=name,
3703            paginator=paginator,
3704            primary_key=primary_key,
3705            requester=requester,
3706            record_selector=record_selector,
3707            stream_slicer=_NO_STREAM_SLICING,
3708            request_option_provider=request_options_provider,
3709            config=config,
3710            ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
3711            additional_query_properties=query_properties,
3712            log_formatter=self._get_log_formatter(log_formatter, name),
3713            pagination_tracker_factory=self._create_pagination_tracker_factory(
3714                model.pagination_reset, cursor
3715            ),
3716            post_pagination_filter=post_pagination_filter,
3717            parameters=model.parameters or {},
3718        )
3719
3720    def _create_pagination_tracker_factory(
3721        self, model: Optional[PaginationResetModel], cursor: Cursor
3722    ) -> Callable[[], PaginationTracker]:
3723        if model is None:
3724            return lambda: PaginationTracker()
3725
3726        # Until we figure out a way to use any cursor for PaginationTracker, we will have to have this cursor selector logic
3727        cursor_factory: Callable[[], Optional[ConcurrentCursor]] = lambda: None
3728        if model.action == PaginationResetActionModel.RESET:
3729            # in that case, we will let cursor_factory to return None even if the stream has a cursor
3730            pass
3731        elif model.action == PaginationResetActionModel.SPLIT_USING_CURSOR:
3732            if isinstance(cursor, ConcurrentCursor):
3733                cursor_factory = lambda: cursor.copy_without_state()  # type: ignore  # the if condition validates that it is a ConcurrentCursor
3734            elif isinstance(cursor, ConcurrentPerPartitionCursor):
3735                cursor_factory = lambda: cursor._cursor_factory.create(  # type: ignore  # if this becomes a problem, we would need to extract the cursor_factory instantiation logic and make it accessible here
3736                    {}, datetime.timedelta(0)
3737                )
3738            elif not isinstance(cursor, FinalStateCursor):
3739                LOGGER.warning(
3740                    "Unknown cursor for PaginationTracker. Pagination resets might not work properly"
3741                )
3742        else:
3743            raise ValueError(f"Unknown PaginationReset action: {model.action}")
3744
3745        limit = model.limits.number_of_records if model and model.limits else None
3746        return lambda: PaginationTracker(cursor_factory(), limit)
3747
3748    def _get_log_formatter(
3749        self, log_formatter: Callable[[Response], Any] | None, name: str
3750    ) -> Callable[[Response], Any] | None:
3751        if self._should_limit_slices_fetched():
3752            return (
3753                (
3754                    lambda response: format_http_message(
3755                        response,
3756                        f"Stream '{name}' request",
3757                        f"Request performed in order to extract records for stream '{name}'",
3758                        name,
3759                    )
3760                )
3761                if not log_formatter
3762                else log_formatter
3763            )
3764        return None
3765
3766    def _should_limit_slices_fetched(self) -> bool:
3767        """
3768        Returns True if the number of slices fetched should be limited, False otherwise.
3769        This is used to limit the number of slices fetched during tests.
3770        """
3771        return bool(self._limit_slices_fetched or self._emit_connector_builder_messages)
3772
3773    @staticmethod
3774    def _has_query_properties_in_request_parameters(
3775        requester: Union[HttpRequesterModel, CustomRequesterModel],
3776    ) -> bool:
3777        if not hasattr(requester, "request_parameters"):
3778            return False
3779        request_parameters = requester.request_parameters
3780        if request_parameters and isinstance(request_parameters, Mapping):
3781            for request_parameter in request_parameters.values():
3782                if isinstance(request_parameter, QueryPropertiesModel):
3783                    return True
3784        return False
3785
3786    @staticmethod
3787    def _remove_query_properties(
3788        request_parameters: Mapping[str, Union[str, QueryPropertiesModel]],
3789    ) -> Mapping[str, str]:
3790        return {
3791            parameter_field: request_parameter
3792            for parameter_field, request_parameter in request_parameters.items()
3793            if not isinstance(request_parameter, QueryPropertiesModel)
3794        }
3795
3796    def create_state_delegating_stream(
3797        self,
3798        model: StateDelegatingStreamModel,
3799        config: Config,
3800        **kwargs: Any,
3801    ) -> DefaultStream:
3802        if (
3803            model.full_refresh_stream.name != model.name
3804            or model.name != model.incremental_stream.name
3805        ):
3806            raise ValueError(
3807                f"state_delegating_stream, full_refresh_stream name and incremental_stream must have equal names. Instead has {model.name}, {model.full_refresh_stream.name} and {model.incremental_stream.name}."
3808            )
3809
3810        # Resolve api_retention_period with config context (supports Jinja2 interpolation)
3811        resolved_retention_period: Optional[str] = None
3812        if model.api_retention_period:
3813            interpolated_retention = InterpolatedString.create(
3814                model.api_retention_period, parameters=model.parameters or {}
3815            )
3816            resolved_value = interpolated_retention.eval(config=config)
3817            if resolved_value:
3818                resolved_retention_period = str(resolved_value)
3819
3820        if resolved_retention_period:
3821            for stream_model in (model.full_refresh_stream, model.incremental_stream):
3822                if isinstance(stream_model.incremental_sync, IncrementingCountCursorModel):
3823                    raise ValueError(
3824                        f"Stream '{model.name}' uses IncrementingCountCursor which is not supported "
3825                        f"with api_retention_period. IncrementingCountCursor does not use datetime-based "
3826                        f"cursors, so cursor age validation cannot be performed."
3827                    )
3828
3829        stream_state = self._connector_state_manager.get_stream_state(model.name, None)
3830
3831        if not stream_state:
3832            return self._create_component_from_model(  # type: ignore[no-any-return]
3833                model.full_refresh_stream, config=config, **kwargs
3834            )
3835
3836        incremental_stream: DefaultStream = self._create_component_from_model(
3837            model.incremental_stream, config=config, **kwargs
3838        )  # type: ignore[assignment]
3839
3840        # Only run cursor age validation for streams that are in the configured
3841        # catalog (or when no catalog was provided, e.g. during discover / connector
3842        # builder).  Streams not selected by the user but instantiated as parent-stream
3843        # dependencies must not go through this path because it emits state messages
3844        # that the destination does not know about, causing "Stream not found" crashes.
3845        stream_is_in_catalog = (
3846            not self._stream_name_to_configured_stream  # no catalog → validate by default
3847            or model.name in self._stream_name_to_configured_stream
3848        )
3849        if resolved_retention_period and stream_is_in_catalog:
3850            full_refresh_stream: DefaultStream = self._create_component_from_model(
3851                model.full_refresh_stream, config=config, **kwargs
3852            )  # type: ignore[assignment]
3853            if self._is_cursor_older_than_retention_period(
3854                stream_state,
3855                full_refresh_stream.cursor,
3856                incremental_stream.cursor,
3857                resolved_retention_period,
3858                model.name,
3859            ):
3860                # Clear state BEFORE constructing the full_refresh_stream so that
3861                # its cursor starts from start_date instead of the stale cursor.
3862                self._connector_state_manager.update_state_for_stream(model.name, None, {})
3863                state_message = self._connector_state_manager.create_state_message(model.name, None)
3864                self._message_repository.emit_message(state_message)
3865                return self._create_component_from_model(  # type: ignore[no-any-return]
3866                    model.full_refresh_stream, config=config, **kwargs
3867                )
3868
3869        return incremental_stream
3870
3871    @staticmethod
3872    def _is_cursor_older_than_retention_period(
3873        stream_state: Mapping[str, Any],
3874        full_refresh_cursor: Cursor,
3875        incremental_cursor: Cursor,
3876        api_retention_period: str,
3877        stream_name: str,
3878    ) -> bool:
3879        """Check if the cursor value in the state is older than the API's retention period.
3880
3881        Checks cursors in sequence: full refresh cursor first, then incremental cursor.
3882        FinalStateCursor returns now() for completed full refresh state (NO_CURSOR_STATE_KEY),
3883        which is always within retention, so we use incremental. For other states, it returns
3884        None and we fall back to checking the incremental cursor.
3885
3886        Returns True if the cursor is older than the retention period (should use full refresh).
3887        Returns False if the cursor is within the retention period (safe to use incremental).
3888        """
3889        retention_duration = parse_duration(api_retention_period)
3890        retention_cutoff = datetime.datetime.now(datetime.timezone.utc) - retention_duration
3891
3892        # Check full refresh cursor first
3893        cursor_datetime = full_refresh_cursor.get_cursor_datetime_from_state(stream_state)
3894
3895        # If full refresh cursor returns None, check incremental cursor
3896        if cursor_datetime is None:
3897            cursor_datetime = incremental_cursor.get_cursor_datetime_from_state(stream_state)
3898
3899        if cursor_datetime is None:
3900            # Neither cursor could parse the state - fall back to full refresh to be safe
3901            return True
3902
3903        if cursor_datetime < retention_cutoff:
3904            logging.warning(
3905                f"Stream '{stream_name}' has a cursor value older than "
3906                f"the API's retention period of {api_retention_period} "
3907                f"(cutoff: {retention_cutoff.isoformat()}). "
3908                f"Falling back to full refresh to avoid data loss."
3909            )
3910            return True
3911
3912        return False
3913
3914    def _get_state_delegating_stream_model(
3915        self,
3916        model: StateDelegatingStreamModel,
3917        parent_state: Optional[Mapping[str, Any]] = None,
3918    ) -> DeclarativeStreamModel:
3919        """Return the appropriate underlying stream model based on state."""
3920        return (
3921            model.incremental_stream
3922            if self._connector_state_manager.get_stream_state(model.name, None) or parent_state
3923            else model.full_refresh_stream
3924        )
3925
3926    _OPTIONAL_ASYNC_STATUS_FIELDS = {"skipped"}
3927
3928    def _create_async_job_status_mapping(
3929        self, model: AsyncJobStatusMapModel, config: Config, **kwargs: Any
3930    ) -> Mapping[str, AsyncJobStatus]:
3931        api_status_to_cdk_status = {}
3932        for cdk_status, api_statuses in model.dict().items():
3933            if cdk_status == "type":
3934                # This is an element of the dict because of the typing of the CDK but it is not a CDK status
3935                continue
3936
3937            if api_statuses is None:
3938                if cdk_status in self._OPTIONAL_ASYNC_STATUS_FIELDS:
3939                    continue
3940                raise ValueError(
3941                    f"Required CDK status '{cdk_status}' has no API statuses mapped. "
3942                    f"Please provide at least an empty list for required status fields."
3943                )
3944
3945            for status in api_statuses:
3946                if status in api_status_to_cdk_status:
3947                    raise ValueError(
3948                        f"API status {status} is already set for CDK status {cdk_status}. Please ensure API statuses are only provided once"
3949                    )
3950                api_status_to_cdk_status[status] = self._get_async_job_status(cdk_status)
3951        return api_status_to_cdk_status
3952
3953    def _get_async_job_status(self, status: str) -> AsyncJobStatus:
3954        match status:
3955            case "running":
3956                return AsyncJobStatus.RUNNING
3957            case "completed":
3958                return AsyncJobStatus.COMPLETED
3959            case "failed":
3960                return AsyncJobStatus.FAILED
3961            case "timeout":
3962                return AsyncJobStatus.TIMED_OUT
3963            case "skipped":
3964                return AsyncJobStatus.SKIPPED
3965            case _:
3966                raise ValueError(f"Unsupported CDK status {status}")
3967
3968    def create_async_retriever(
3969        self,
3970        model: AsyncRetrieverModel,
3971        config: Config,
3972        *,
3973        name: str,
3974        primary_key: Optional[
3975            Union[str, List[str], List[List[str]]]
3976        ],  # this seems to be needed to match create_simple_retriever
3977        stream_slicer: Optional[StreamSlicer],
3978        client_side_incremental_sync: Optional[Dict[str, Any]] = None,
3979        transformations: List[RecordTransformation],
3980        **kwargs: Any,
3981    ) -> AsyncRetriever:
3982        if model.download_target_requester and not model.download_target_extractor:
3983            raise ValueError(
3984                f"`download_target_extractor` required if using a `download_target_requester`"
3985            )
3986
3987        def _get_download_retriever(
3988            requester: Requester, extractor: RecordExtractor, _decoder: Decoder
3989        ) -> SimpleRetriever:
3990            # We create a record selector for the download retriever
3991            # with no schema normalization and no transformations, neither record filter
3992            # as all this occurs in the record_selector of the AsyncRetriever
3993            record_selector = RecordSelector(
3994                extractor=extractor,
3995                name=name,
3996                record_filter=None,
3997                transformations=[],
3998                schema_normalization=TypeTransformer(TransformConfig.NoTransform),
3999                config=config,
4000                parameters={},
4001            )
4002            paginator = (
4003                self._create_component_from_model(
4004                    model=model.download_paginator,
4005                    decoder=_decoder,
4006                    config=config,
4007                    url_base="",
4008                )
4009                if model.download_paginator
4010                else NoPagination(parameters={})
4011            )
4012
4013            return SimpleRetriever(
4014                requester=requester,
4015                record_selector=record_selector,
4016                primary_key=None,
4017                name=name,
4018                paginator=paginator,
4019                config=config,
4020                parameters={},
4021                log_formatter=self._get_log_formatter(None, name),
4022            )
4023
4024        def _get_job_timeout() -> datetime.timedelta:
4025            user_defined_timeout: Optional[int] = (
4026                int(
4027                    InterpolatedString.create(
4028                        str(model.polling_job_timeout),
4029                        parameters={},
4030                    ).eval(config)
4031                )
4032                if model.polling_job_timeout
4033                else None
4034            )
4035
4036            # check for user defined timeout during the test read or 15 minutes
4037            test_read_timeout = datetime.timedelta(minutes=user_defined_timeout or 15)
4038            # default value for non-connector builder is 60 minutes.
4039            default_sync_timeout = datetime.timedelta(minutes=user_defined_timeout or 60)
4040
4041            return (
4042                test_read_timeout if self._emit_connector_builder_messages else default_sync_timeout
4043            )
4044
4045        decoder = (
4046            self._create_component_from_model(model=model.decoder, config=config)
4047            if model.decoder
4048            else JsonDecoder(parameters={})
4049        )
4050        record_selector = self._create_component_from_model(
4051            model=model.record_selector,
4052            config=config,
4053            decoder=decoder,
4054            name=name,
4055            transformations=transformations,
4056            client_side_incremental_sync=client_side_incremental_sync,
4057        )
4058
4059        stream_slicer = stream_slicer or SinglePartitionRouter(parameters={})
4060        if self._should_limit_slices_fetched():
4061            stream_slicer = cast(
4062                StreamSlicer,
4063                StreamSlicerTestReadDecorator(
4064                    wrapped_slicer=stream_slicer,
4065                    maximum_number_of_slices=self._limit_slices_fetched or 5,
4066                ),
4067            )
4068
4069        creation_requester = self._create_component_from_model(
4070            model=model.creation_requester,
4071            decoder=decoder,
4072            config=config,
4073            name=f"job creation - {name}",
4074        )
4075        polling_requester = self._create_component_from_model(
4076            model=model.polling_requester,
4077            decoder=decoder,
4078            config=config,
4079            name=f"job polling - {name}",
4080        )
4081        job_download_components_name = f"job download - {name}"
4082        download_decoder = (
4083            self._create_component_from_model(model=model.download_decoder, config=config)
4084            if model.download_decoder
4085            else JsonDecoder(parameters={})
4086        )
4087        download_extractor = (
4088            self._create_component_from_model(
4089                model=model.download_extractor,
4090                config=config,
4091                decoder=download_decoder,
4092                parameters=model.parameters,
4093            )
4094            if model.download_extractor
4095            else DpathExtractor(
4096                [],
4097                config=config,
4098                decoder=download_decoder,
4099                parameters=model.parameters or {},
4100            )
4101        )
4102        download_requester = self._create_component_from_model(
4103            model=model.download_requester,
4104            decoder=download_decoder,
4105            config=config,
4106            name=job_download_components_name,
4107        )
4108        download_retriever = _get_download_retriever(
4109            download_requester, download_extractor, download_decoder
4110        )
4111        abort_requester = (
4112            self._create_component_from_model(
4113                model=model.abort_requester,
4114                decoder=decoder,
4115                config=config,
4116                name=f"job abort - {name}",
4117            )
4118            if model.abort_requester
4119            else None
4120        )
4121        delete_requester = (
4122            self._create_component_from_model(
4123                model=model.delete_requester,
4124                decoder=decoder,
4125                config=config,
4126                name=f"job delete - {name}",
4127            )
4128            if model.delete_requester
4129            else None
4130        )
4131        download_target_requester = (
4132            self._create_component_from_model(
4133                model=model.download_target_requester,
4134                decoder=decoder,
4135                config=config,
4136                name=f"job extract_url - {name}",
4137            )
4138            if model.download_target_requester
4139            else None
4140        )
4141        status_extractor = self._create_component_from_model(
4142            model=model.status_extractor, decoder=decoder, config=config, name=name
4143        )
4144        download_target_extractor = (
4145            self._create_component_from_model(
4146                model=model.download_target_extractor,
4147                decoder=decoder,
4148                config=config,
4149                name=name,
4150            )
4151            if model.download_target_extractor
4152            else None
4153        )
4154
4155        job_repository: AsyncJobRepository = AsyncHttpJobRepository(
4156            creation_requester=creation_requester,
4157            polling_requester=polling_requester,
4158            download_retriever=download_retriever,
4159            download_target_requester=download_target_requester,
4160            abort_requester=abort_requester,
4161            delete_requester=delete_requester,
4162            status_extractor=status_extractor,
4163            status_mapping=self._create_async_job_status_mapping(model.status_mapping, config),
4164            download_target_extractor=download_target_extractor,
4165            job_timeout=_get_job_timeout(),
4166        )
4167
4168        failed_retry_wait_time_in_seconds: Optional[int] = (
4169            int(
4170                InterpolatedString.create(
4171                    str(model.failed_retry_wait_time_in_seconds),
4172                    parameters={},
4173                ).eval(config)
4174            )
4175            if model.failed_retry_wait_time_in_seconds
4176            else None
4177        )
4178
4179        async_job_partition_router = AsyncJobPartitionRouter(
4180            job_orchestrator_factory=lambda stream_slices: AsyncJobOrchestrator(
4181                job_repository,
4182                stream_slices,
4183                self._job_tracker,
4184                self._message_repository,
4185                # FIXME work would need to be done here in order to detect if a stream as a parent stream that is bulk
4186                has_bulk_parent=False,
4187                # set the `job_max_retry` to 1 for the `Connector Builder`` use-case.
4188                # `None` == default retry is set to 3 attempts, under the hood.
4189                job_max_retry=1 if self._emit_connector_builder_messages else None,
4190                failed_retry_wait_time_in_seconds=failed_retry_wait_time_in_seconds,
4191            ),
4192            stream_slicer=stream_slicer,
4193            config=config,
4194            parameters=model.parameters or {},
4195        )
4196
4197        return AsyncRetriever(
4198            record_selector=record_selector,
4199            stream_slicer=async_job_partition_router,
4200            config=config,
4201            parameters=model.parameters or {},
4202        )
4203
4204    def create_spec(self, model: SpecModel, config: Config, **kwargs: Any) -> Spec:
4205        config_migrations = [
4206            self._create_component_from_model(migration, config)
4207            for migration in (
4208                model.config_normalization_rules.config_migrations
4209                if (
4210                    model.config_normalization_rules
4211                    and model.config_normalization_rules.config_migrations
4212                )
4213                else []
4214            )
4215        ]
4216        config_transformations = [
4217            self._create_component_from_model(transformation, config)
4218            for transformation in (
4219                model.config_normalization_rules.transformations
4220                if (
4221                    model.config_normalization_rules
4222                    and model.config_normalization_rules.transformations
4223                )
4224                else []
4225            )
4226        ]
4227        config_validations = [
4228            self._create_component_from_model(validation, config)
4229            for validation in (
4230                model.config_normalization_rules.validations
4231                if (
4232                    model.config_normalization_rules
4233                    and model.config_normalization_rules.validations
4234                )
4235                else []
4236            )
4237        ]
4238
4239        return Spec(
4240            connection_specification=model.connection_specification,
4241            documentation_url=model.documentation_url,
4242            advanced_auth=model.advanced_auth,
4243            parameters={},
4244            config_migrations=config_migrations,
4245            config_transformations=config_transformations,
4246            config_validations=config_validations,
4247        )
4248
4249    def create_substream_partition_router(
4250        self,
4251        model: SubstreamPartitionRouterModel,
4252        config: Config,
4253        *,
4254        stream_name: str,
4255        **kwargs: Any,
4256    ) -> SubstreamPartitionRouter:
4257        parent_stream_configs = []
4258        if model.parent_stream_configs:
4259            parent_stream_configs.extend(
4260                [
4261                    self.create_parent_stream_config_with_substream_wrapper(
4262                        model=parent_stream_config, config=config, stream_name=stream_name, **kwargs
4263                    )
4264                    for parent_stream_config in model.parent_stream_configs
4265                ]
4266            )
4267
4268        return SubstreamPartitionRouter(
4269            parent_stream_configs=parent_stream_configs,
4270            parameters=model.parameters or {},
4271            config=config,
4272        )
4273
4274    def create_parent_stream_config_with_substream_wrapper(
4275        self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any
4276    ) -> Any:
4277        child_state = self._connector_state_manager.get_stream_state(stream_name, None)
4278        if NO_CURSOR_STATE_KEY in child_state:
4279            # Full refresh streams checkpoint a `{NO_CURSOR_STATE_KEY: true}` sentinel. When such a
4280            # stream is later converted to incremental with an incremental_dependency parent,
4281            # `_instantiate_parent_stream_state_manager` would treat the sentinel's boolean as a legacy
4282            # cursor value and re-key it under the parent's cursor field, crashing cursor initialization.
4283            child_state = {
4284                key: value for key, value in child_state.items() if key != NO_CURSOR_STATE_KEY
4285            }
4286
4287        parent_state: Optional[Mapping[str, Any]] = (
4288            child_state if model.incremental_dependency and child_state else None
4289        )
4290        connector_state_manager = self._instantiate_parent_stream_state_manager(
4291            child_state, config, model, parent_state
4292        )
4293
4294        substream_factory = ModelToComponentFactory(
4295            custom_components_trusted=self._custom_components_trusted,
4296            connector_state_manager=connector_state_manager,
4297            limit_pages_fetched_per_slice=self._limit_pages_fetched_per_slice,
4298            limit_slices_fetched=self._limit_slices_fetched,
4299            emit_connector_builder_messages=self._emit_connector_builder_messages,
4300            disable_retries=self._disable_retries,
4301            disable_cache=self._disable_cache,
4302            message_repository=StateFilteringMessageRepository(
4303                LogAppenderMessageRepositoryDecorator(
4304                    {
4305                        "airbyte_cdk": {"stream": {"is_substream": True}},
4306                        "http": {"is_auxiliary": True},
4307                    },
4308                    self._message_repository,
4309                    self._evaluate_log_level(self._emit_connector_builder_messages),
4310                ),
4311            ),
4312            api_budget=self._api_budget,
4313            # Share the authenticator registry so parent and child streams draw from the
4314            # same token quota counters
4315            rate_limited_authenticators=self._rate_limited_authenticators,
4316        )
4317
4318        return substream_factory.create_parent_stream_config(
4319            model=model, config=config, stream_name=stream_name, **kwargs
4320        )
4321
4322    def _instantiate_parent_stream_state_manager(
4323        self,
4324        child_state: MutableMapping[str, Any],
4325        config: Config,
4326        model: ParentStreamConfigModel,
4327        parent_state: Optional[Mapping[str, Any]] = None,
4328    ) -> ConnectorStateManager:
4329        """
4330        With DefaultStream, the state needs to be provided during __init__ of the cursor as opposed to the
4331        `set_initial_state` flow that existed for the declarative cursors. This state is taken from
4332        self._connector_state_manager.get_stream_state (`self` being a newly created ModelToComponentFactory to account
4333        for the MessageRepository being different). So we need to pass a ConnectorStateManager to the
4334        ModelToComponentFactory that has the parent states. This method populates this if there is a child state and if
4335        incremental_dependency is set.
4336        """
4337        if model.incremental_dependency and child_state:
4338            parent_stream_name = model.stream.name or ""
4339            extracted_parent_state = ConcurrentPerPartitionCursor.get_parent_state(
4340                child_state, parent_stream_name
4341            )
4342
4343            if not extracted_parent_state:
4344                extracted_parent_state = ConcurrentPerPartitionCursor.get_global_state(
4345                    child_state, parent_stream_name
4346                )
4347
4348                if not extracted_parent_state and not isinstance(extracted_parent_state, dict):
4349                    cursor_values = child_state.values()
4350                    if cursor_values and len(cursor_values) == 1:
4351                        incremental_sync_model: Union[
4352                            DatetimeBasedCursorModel,
4353                            IncrementingCountCursorModel,
4354                        ] = (
4355                            model.stream.incremental_sync  # type: ignore  # if we are there, it is because there is incremental_dependency and therefore there is an incremental_sync on the parent stream
4356                            if isinstance(model.stream, DeclarativeStreamModel)
4357                            else self._get_state_delegating_stream_model(
4358                                model.stream, parent_state=parent_state
4359                            ).incremental_sync
4360                        )
4361                        cursor_field = InterpolatedString.create(
4362                            incremental_sync_model.cursor_field,
4363                            parameters=incremental_sync_model.parameters or {},
4364                        ).eval(config)
4365                        extracted_parent_state = AirbyteStateMessage(
4366                            type=AirbyteStateType.STREAM,
4367                            stream=AirbyteStreamState(
4368                                stream_descriptor=StreamDescriptor(
4369                                    name=parent_stream_name, namespace=None
4370                                ),
4371                                stream_state=AirbyteStateBlob(
4372                                    {cursor_field: list(cursor_values)[0]}
4373                                ),
4374                            ),
4375                        )
4376            return ConnectorStateManager([extracted_parent_state] if extracted_parent_state else [])
4377
4378        return ConnectorStateManager([])
4379
4380    @staticmethod
4381    def create_wait_time_from_header(
4382        model: WaitTimeFromHeaderModel, config: Config, **kwargs: Any
4383    ) -> WaitTimeFromHeaderBackoffStrategy:
4384        return WaitTimeFromHeaderBackoffStrategy(
4385            header=model.header,
4386            parameters=model.parameters or {},
4387            config=config,
4388            regex=model.regex,
4389            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
4390        )
4391
4392    @staticmethod
4393    def create_wait_until_time_from_header(
4394        model: WaitUntilTimeFromHeaderModel, config: Config, **kwargs: Any
4395    ) -> WaitUntilTimeFromHeaderBackoffStrategy:
4396        return WaitUntilTimeFromHeaderBackoffStrategy(
4397            header=model.header,
4398            parameters=model.parameters or {},
4399            config=config,
4400            min_wait=model.min_wait,
4401            regex=model.regex,
4402            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
4403        )
4404
4405    def get_message_repository(self) -> MessageRepository:
4406        return self._message_repository
4407
4408    def _evaluate_log_level(self, emit_connector_builder_messages: bool) -> Level:
4409        return Level.DEBUG if emit_connector_builder_messages else Level.INFO
4410
4411    @staticmethod
4412    def create_components_mapping_definition(
4413        model: ComponentMappingDefinitionModel, config: Config, **kwargs: Any
4414    ) -> ComponentMappingDefinition:
4415        interpolated_value = InterpolatedString.create(
4416            model.value, parameters=model.parameters or {}
4417        )
4418        field_path = [
4419            InterpolatedString.create(path, parameters=model.parameters or {})
4420            for path in model.field_path
4421        ]
4422        return ComponentMappingDefinition(
4423            field_path=field_path,  # type: ignore[arg-type] # field_path can be str and InterpolatedString
4424            value=interpolated_value,
4425            value_type=ModelToComponentFactory._json_schema_type_name_to_type(model.value_type),
4426            create_or_update=model.create_or_update,
4427            condition=model.condition,
4428            parameters=model.parameters or {},
4429        )
4430
4431    def create_http_components_resolver(
4432        self, model: HttpComponentsResolverModel, config: Config, stream_name: Optional[str] = None
4433    ) -> Any:
4434        retriever = self._create_component_from_model(
4435            model=model.retriever,
4436            config=config,
4437            name=f"{stream_name if stream_name else '__http_components_resolver'}",
4438            primary_key=None,
4439            stream_slicer=self._build_stream_slicer_from_partition_router(model.retriever, config),
4440            transformations=[],
4441        )
4442
4443        components_mapping = []
4444        for component_mapping_definition_model in model.components_mapping:
4445            if component_mapping_definition_model.condition:
4446                raise ValueError("`condition` is only supported for     `ConfigComponentsResolver`")
4447            components_mapping.append(
4448                self._create_component_from_model(
4449                    model=component_mapping_definition_model,
4450                    value_type=ModelToComponentFactory._json_schema_type_name_to_type(
4451                        component_mapping_definition_model.value_type
4452                    ),
4453                    config=config,
4454                )
4455            )
4456
4457        return HttpComponentsResolver(
4458            retriever=retriever,
4459            stream_slicer=self._build_stream_slicer_from_partition_router(model.retriever, config),
4460            config=config,
4461            components_mapping=components_mapping,
4462            parameters=model.parameters or {},
4463        )
4464
4465    @staticmethod
4466    def create_stream_config(
4467        model: StreamConfigModel, config: Config, **kwargs: Any
4468    ) -> StreamConfig:
4469        model_configs_pointer: List[Union[InterpolatedString, str]] = (
4470            [x for x in model.configs_pointer] if model.configs_pointer else []
4471        )
4472
4473        return StreamConfig(
4474            configs_pointer=model_configs_pointer,
4475            default_values=model.default_values,
4476            parameters=model.parameters or {},
4477        )
4478
4479    def create_config_components_resolver(
4480        self,
4481        model: ConfigComponentsResolverModel,
4482        config: Config,
4483    ) -> Any:
4484        model_stream_configs = (
4485            model.stream_config if isinstance(model.stream_config, list) else [model.stream_config]
4486        )
4487
4488        stream_configs = [
4489            self._create_component_from_model(
4490                stream_config, config=config, parameters=model.parameters or {}
4491            )
4492            for stream_config in model_stream_configs
4493        ]
4494
4495        components_mapping = [
4496            self._create_component_from_model(
4497                model=components_mapping_definition_model,
4498                value_type=ModelToComponentFactory._json_schema_type_name_to_type(
4499                    components_mapping_definition_model.value_type
4500                ),
4501                config=config,
4502                parameters=model.parameters,
4503            )
4504            for components_mapping_definition_model in model.components_mapping
4505        ]
4506
4507        return ConfigComponentsResolver(
4508            stream_configs=stream_configs,
4509            config=config,
4510            components_mapping=components_mapping,
4511            parameters=model.parameters or {},
4512        )
4513
4514    def create_parametrized_components_resolver(
4515        self,
4516        model: ParametrizedComponentsResolverModel,
4517        config: Config,
4518    ) -> ParametrizedComponentsResolver:
4519        stream_parameters = StreamParametersDefinition(
4520            list_of_parameters_for_stream=model.stream_parameters.list_of_parameters_for_stream
4521        )
4522
4523        components_mapping = []
4524        for components_mapping_definition_model in model.components_mapping:
4525            if components_mapping_definition_model.condition:
4526                raise ValueError("`condition` is only supported for `ConfigComponentsResolver`")
4527            components_mapping.append(
4528                self._create_component_from_model(
4529                    model=components_mapping_definition_model,
4530                    value_type=ModelToComponentFactory._json_schema_type_name_to_type(
4531                        components_mapping_definition_model.value_type
4532                    ),
4533                    config=config,
4534                )
4535            )
4536        return ParametrizedComponentsResolver(
4537            stream_parameters=stream_parameters,
4538            config=config,
4539            components_mapping=components_mapping,
4540            parameters=model.parameters or {},
4541        )
4542
4543    _UNSUPPORTED_DECODER_ERROR = (
4544        "Specified decoder of {decoder_type} is not supported for pagination."
4545        "Please set as `JsonDecoder`, `XmlDecoder`, or a `CompositeRawDecoder` with an inner_parser of `JsonParser` or `GzipParser` instead."
4546        "If using `GzipParser`, please ensure that the lowest level inner_parser is a `JsonParser`."
4547    )
4548
4549    def _is_supported_decoder_for_pagination(self, decoder: Decoder) -> bool:
4550        if isinstance(decoder, (JsonDecoder, XmlDecoder)):
4551            return True
4552        elif isinstance(decoder, CompositeRawDecoder):
4553            return self._is_supported_parser_for_pagination(decoder.parser)
4554        else:
4555            return False
4556
4557    def _is_supported_parser_for_pagination(self, parser: Parser) -> bool:
4558        if isinstance(parser, JsonParser):
4559            return True
4560        elif isinstance(parser, GzipParser):
4561            return isinstance(parser.inner_parser, JsonParser)
4562        else:
4563            return False
4564
4565    def create_http_api_budget(
4566        self, model: HTTPAPIBudgetModel, config: Config, **kwargs: Any
4567    ) -> HttpAPIBudget:
4568        policies = [
4569            self._create_component_from_model(model=policy, config=config)
4570            for policy in model.policies
4571        ]
4572
4573        return HttpAPIBudget(
4574            policies=policies,
4575            ratelimit_reset_header=model.ratelimit_reset_header or "ratelimit-reset",
4576            ratelimit_remaining_header=model.ratelimit_remaining_header or "ratelimit-remaining",
4577            status_codes_for_ratelimit_hit=model.status_codes_for_ratelimit_hit or [429],
4578        )
4579
4580    def create_fixed_window_call_rate_policy(
4581        self, model: FixedWindowCallRatePolicyModel, config: Config, **kwargs: Any
4582    ) -> FixedWindowCallRatePolicy:
4583        matchers = [
4584            self._create_component_from_model(model=matcher, config=config)
4585            for matcher in model.matchers
4586        ]
4587
4588        # Set the initial reset timestamp to 10 days from now.
4589        # This value will be updated by the first request.
4590        return FixedWindowCallRatePolicy(
4591            next_reset_ts=datetime.datetime.now() + datetime.timedelta(days=10),
4592            period=parse_duration(model.period),
4593            call_limit=model.call_limit,
4594            matchers=matchers,
4595        )
4596
4597    def create_file_uploader(
4598        self, model: FileUploaderModel, config: Config, **kwargs: Any
4599    ) -> FileUploader:
4600        name = "File Uploader"
4601        requester = self._create_component_from_model(
4602            model=model.requester,
4603            config=config,
4604            name=name,
4605            **kwargs,
4606        )
4607        download_target_extractor = self._create_component_from_model(
4608            model=model.download_target_extractor,
4609            config=config,
4610            name=name,
4611            **kwargs,
4612        )
4613        emit_connector_builder_messages = self._emit_connector_builder_messages
4614        file_uploader = DefaultFileUploader(
4615            requester=requester,
4616            download_target_extractor=download_target_extractor,
4617            config=config,
4618            file_writer=NoopFileWriter()
4619            if emit_connector_builder_messages
4620            else LocalFileSystemFileWriter(),
4621            parameters=model.parameters or {},
4622            filename_extractor=model.filename_extractor if model.filename_extractor else None,
4623        )
4624
4625        return (
4626            ConnectorBuilderFileUploader(file_uploader)
4627            if emit_connector_builder_messages
4628            else file_uploader
4629        )
4630
4631    def create_moving_window_call_rate_policy(
4632        self, model: MovingWindowCallRatePolicyModel, config: Config, **kwargs: Any
4633    ) -> MovingWindowCallRatePolicy:
4634        rates = [
4635            self._create_component_from_model(model=rate, config=config) for rate in model.rates
4636        ]
4637        matchers = [
4638            self._create_component_from_model(model=matcher, config=config)
4639            for matcher in model.matchers
4640        ]
4641        return MovingWindowCallRatePolicy(
4642            rates=rates,
4643            matchers=matchers,
4644        )
4645
4646    def create_unlimited_call_rate_policy(
4647        self, model: UnlimitedCallRatePolicyModel, config: Config, **kwargs: Any
4648    ) -> UnlimitedCallRatePolicy:
4649        matchers = [
4650            self._create_component_from_model(model=matcher, config=config)
4651            for matcher in model.matchers
4652        ]
4653
4654        return UnlimitedCallRatePolicy(
4655            matchers=matchers,
4656        )
4657
4658    def create_rate(self, model: RateModel, config: Config, **kwargs: Any) -> Rate:
4659        interpolated_limit = InterpolatedString.create(str(model.limit), parameters={})
4660        return Rate(
4661            limit=int(interpolated_limit.eval(config=config)),
4662            interval=parse_duration(model.interval),
4663        )
4664
4665    def create_http_request_matcher(
4666        self, model: HttpRequestRegexMatcherModel, config: Config, **kwargs: Any
4667    ) -> HttpRequestRegexMatcher:
4668        weight = model.weight
4669        if weight is not None:
4670            if isinstance(weight, str):
4671                weight = int(InterpolatedString.create(weight, parameters={}).eval(config))
4672            else:
4673                weight = int(weight)
4674            if weight < 1:
4675                raise ValueError(f"weight must be >= 1, got {weight}")
4676        return HttpRequestRegexMatcher(
4677            method=model.method,
4678            url_base=model.url_base,
4679            url_path_pattern=model.url_path_pattern,
4680            params=model.params,
4681            headers=model.headers,
4682            weight=weight,
4683        )
4684
4685    def create_rate_limited_multiple_token_authenticator(
4686        self,
4687        model: RateLimitedMultipleTokenAuthenticatorModel,
4688        config: Config,
4689        **kwargs: Any,
4690    ) -> RateLimitedMultipleTokenAuthenticator:
4691        if isinstance(model.tokens, str):
4692            tokens_value = InterpolatedString.create(model.tokens, parameters={}).eval(config)
4693            delimiter = model.token_delimiter or ","
4694            tokens = [
4695                token.strip() for token in str(tokens_value).split(delimiter) if token.strip()
4696            ]
4697        else:
4698            tokens = [
4699                token_value
4700                for token in model.tokens
4701                if (
4702                    token_value := str(
4703                        InterpolatedString.create(token, parameters={}).eval(config)
4704                    ).strip()
4705                )
4706            ]
4707
4708        quota_specs = [
4709            {
4710                "name": quota_model.name,
4711                "remaining_path": quota_model.remaining_path,
4712                "reset_path": quota_model.reset_path,
4713                "limit_path": quota_model.limit_path,
4714                "remaining_header": quota_model.remaining_header,
4715                "reset_header": quota_model.reset_header,
4716                "limit_header": quota_model.limit_header,
4717                # Normalize the same way as the runtime TokenQuota below, so an omitted field
4718                # and an explicit `[]` key identically and keep sharing one set of counters.
4719                "exhaustion_status_codes": quota_model.exhaustion_status_codes or [],
4720                "matchers": [
4721                    {
4722                        "method": matcher_model.method,
4723                        "url_base": matcher_model.url_base,
4724                        "url_path_pattern": matcher_model.url_path_pattern,
4725                        "params": matcher_model.params,
4726                        "headers": matcher_model.headers,
4727                        "weight": matcher_model.weight,
4728                    }
4729                    for matcher_model in quota_model.matchers or []
4730                ],
4731            }
4732            for quota_model in model.quotas
4733        ]
4734
4735        quota_status_url = str(
4736            InterpolatedString.create(model.quota_status_source.url, parameters={}).eval(config)
4737        )
4738        quota_status_http_method = (
4739            model.quota_status_source.http_method.value
4740            if model.quota_status_source.http_method
4741            else "GET"
4742        )
4743        quota_status_headers = {
4744            key: str(InterpolatedString.create(value, parameters={}).eval(config))
4745            for key, value in (model.quota_status_source.request_headers or {}).items()
4746        }
4747        # Normalize the same way as the quota specs above, so an omitted field and an explicit
4748        # `[]` key identically and keep sharing one set of counters. Deduplicated as well as
4749        # sorted, because the runtime turns this into a set: without it `[404]` and `[404, 404]`
4750        # would key differently and stop sharing counters while behaving identically.
4751        quota_status_unavailable_status_codes = sorted(
4752            set(model.quota_status_source.unavailable_status_codes or [])
4753        )
4754        auth_method = model.auth_method or "Bearer"
4755        header = model.header or "Authorization"
4756        max_wait_time_str = str(
4757            InterpolatedString.create(model.max_wait_time or "PT2H", parameters={}).eval(config)
4758        )
4759        max_wait_time = parse_duration(max_wait_time_str)
4760        if not isinstance(max_wait_time, datetime.timedelta):
4761            raise ValueError(
4762                f"max_wait_time must be a fixed-length ISO 8601 duration (e.g. 'PT2H'); "
4763                f"calendar-unit durations like '{max_wait_time_str}' are not supported"
4764            )
4765        budget_reserve_fraction = (
4766            model.budget_reserve_fraction if model.budget_reserve_fraction is not None else 0.1
4767        )
4768        budget_min_reserve = (
4769            model.budget_min_reserve if model.budget_min_reserve is not None else 50
4770        )
4771
4772        # Reuse the same instance for identical definitions so that all streams share the
4773        # same token quota counters (similar to how api_budget is shared). The key is built
4774        # from the resolved constructor arguments rather than the raw model so that
4775        # stream-specific `$parameters` propagated onto the model (and its nested components)
4776        # cannot break instance sharing.
4777        cache_key = json.dumps(
4778            {
4779                "tokens": tokens,
4780                "quotas": quota_specs,
4781                "quota_status_url": quota_status_url,
4782                "quota_status_http_method": quota_status_http_method,
4783                "quota_status_headers": quota_status_headers,
4784                "quota_status_unavailable_status_codes": quota_status_unavailable_status_codes,
4785                "auth_method": auth_method,
4786                "header": header,
4787                "max_wait_time": max_wait_time.total_seconds(),
4788                "budget_reserve_fraction": budget_reserve_fraction,
4789                "budget_min_reserve": budget_min_reserve,
4790            },
4791            sort_keys=True,
4792        )
4793        if cache_key in self._rate_limited_authenticators:
4794            return self._rate_limited_authenticators[cache_key]
4795
4796        quotas = [
4797            TokenQuota(
4798                name=quota_model.name,
4799                remaining_path=quota_model.remaining_path,
4800                reset_path=quota_model.reset_path,
4801                limit_path=quota_model.limit_path,
4802                remaining_header=quota_model.remaining_header,
4803                reset_header=quota_model.reset_header,
4804                limit_header=quota_model.limit_header,
4805                exhaustion_status_codes=quota_model.exhaustion_status_codes or [],
4806                matchers=[
4807                    self.create_http_request_matcher(matcher_model, config)
4808                    for matcher_model in quota_model.matchers or []
4809                ],
4810            )
4811            for quota_model in model.quotas
4812        ]
4813
4814        authenticator = RateLimitedMultipleTokenAuthenticator(
4815            tokens=tokens,
4816            quotas=quotas,
4817            quota_status_url=quota_status_url,
4818            quota_status_http_method=quota_status_http_method,
4819            quota_status_headers=quota_status_headers,
4820            quota_status_unavailable_status_codes=quota_status_unavailable_status_codes,
4821            auth_method=auth_method,
4822            header=header,
4823            max_wait_time=max_wait_time,
4824            budget_reserve_fraction=budget_reserve_fraction,
4825            budget_min_reserve=budget_min_reserve,
4826        )
4827        self._rate_limited_authenticators[cache_key] = authenticator
4828        return authenticator
4829
4830    def set_api_budget(self, component_definition: ComponentDefinition, config: Config) -> None:
4831        self._api_budget = self.create_component(
4832            model_type=HTTPAPIBudgetModel, component_definition=component_definition, config=config
4833        )
4834
4835    def create_grouping_partition_router(
4836        self,
4837        model: GroupingPartitionRouterModel,
4838        config: Config,
4839        *,
4840        stream_name: str,
4841        **kwargs: Any,
4842    ) -> GroupingPartitionRouter:
4843        underlying_router = self._create_component_from_model(
4844            model=model.underlying_partition_router,
4845            config=config,
4846            stream_name=stream_name,
4847            **kwargs,
4848        )
4849        if model.group_size < 1:
4850            raise ValueError(f"Group size must be greater than 0, got {model.group_size}")
4851
4852        # Request options in underlying partition routers are not supported for GroupingPartitionRouter
4853        # because they are specific to individual partitions and cannot be aggregated or handled
4854        # when grouping, potentially leading to incorrect API calls. Any request customization
4855        # should be managed at the stream level through the requester's configuration.
4856        if isinstance(underlying_router, SubstreamPartitionRouter):
4857            if any(
4858                parent_config.request_option
4859                for parent_config in underlying_router.parent_stream_configs
4860            ):
4861                raise ValueError("Request options are not supported for GroupingPartitionRouter.")
4862
4863        if isinstance(underlying_router, ListPartitionRouter):
4864            if underlying_router.request_option:
4865                raise ValueError("Request options are not supported for GroupingPartitionRouter.")
4866
4867        return GroupingPartitionRouter(
4868            group_size=model.group_size,
4869            underlying_partition_router=underlying_router,
4870            deduplicate=model.deduplicate if model.deduplicate is not None else True,
4871            config=config,
4872        )
4873
4874    def create_union_partition_router(
4875        self,
4876        model: UnionPartitionRouterModel,
4877        config: Config,
4878        *,
4879        stream_name: str,
4880        **kwargs: Any,
4881    ) -> UnionPartitionRouter:
4882        # The schema enforces minItems: 2 for manifests; this guard covers construction paths
4883        # that bypass JSON-schema validation (the generated model carries no min_items constraint).
4884        if len(model.partition_routers) < 2:
4885            raise ValueError(
4886                f"UnionPartitionRouter for stream {stream_name} needs at least 2 child partition routers"
4887            )
4888
4889        partition_routers = [
4890            self._create_component_from_model(
4891                model=child,
4892                config=config,
4893                stream_name=stream_name,
4894                **kwargs,
4895            )
4896            for child in model.partition_routers
4897        ]
4898
4899        # partition_field depends only on config/parameters, so it is evaluated once at build
4900        # time; the runtime component always receives a plain string.
4901        partition_field = InterpolatedString.create(
4902            model.partition_field, parameters=model.parameters or {}
4903        ).eval(config)
4904
4905        # Fail fast at build time when a built-in child router is statically known to emit a
4906        # partition field different from the union's. CustomPartitionRouter children are opaque
4907        # and can only be validated at runtime.
4908        for child_model in model.partition_routers:
4909            child_partition_fields: List[str] = []
4910            if isinstance(child_model, ListPartitionRouterModel):
4911                child_partition_fields.append(
4912                    InterpolatedString.create(
4913                        child_model.cursor_field, parameters=child_model.parameters or {}
4914                    ).eval(config)
4915                )
4916            elif isinstance(child_model, SubstreamPartitionRouterModel):
4917                for parent_stream_config in child_model.parent_stream_configs:
4918                    child_partition_fields.append(
4919                        InterpolatedString.create(
4920                            parent_stream_config.partition_field,
4921                            parameters=parent_stream_config.parameters
4922                            or child_model.parameters
4923                            or {},
4924                        ).eval(config)
4925                    )
4926            elif isinstance(child_model, UnionPartitionRouterModel):
4927                child_partition_fields.append(
4928                    InterpolatedString.create(
4929                        child_model.partition_field, parameters=child_model.parameters or {}
4930                    ).eval(config)
4931                )
4932            for child_partition_field in child_partition_fields:
4933                if child_partition_field != partition_field:
4934                    raise ValueError(
4935                        f"UnionPartitionRouter expects all child partition routers to emit the "
4936                        f"partition field '{partition_field}', but a "
4937                        f"{child_model.type} child emits '{child_partition_field}'."
4938                    )
4939
4940        # A union slice comes from exactly one child partition router, so request options
4941        # declared on children cannot be applied consistently to requests built from the
4942        # normalized union slices. Partition values should be consumed via interpolation
4943        # (e.g. stream_partition) instead. Note that this validation only covers built-in
4944        # router types; CustomPartitionRouter children are opaque, so any request options
4945        # they implement internally cannot be detected or rejected here.
4946        for router in partition_routers:
4947            if isinstance(router, SubstreamPartitionRouter):
4948                if any(
4949                    parent_config.request_option for parent_config in router.parent_stream_configs
4950                ):
4951                    raise ValueError("Request options are not supported for UnionPartitionRouter.")
4952            if isinstance(router, ListPartitionRouter) and router.request_option:
4953                raise ValueError("Request options are not supported for UnionPartitionRouter.")
4954
4955        return UnionPartitionRouter(
4956            partition_routers=partition_routers,
4957            partition_field=partition_field,
4958            parameters=model.parameters or {},
4959        )
4960
4961    def _ensure_query_properties_to_model(
4962        self, requester: Union[HttpRequesterModel, CustomRequesterModel]
4963    ) -> None:
4964        """
4965        For some reason, it seems like CustomRequesterModel request_parameters stays as dictionaries which means that
4966        the other conditions relying on it being QueryPropertiesModel instead of a dict fail. Here, we migrate them to
4967        proper model.
4968        """
4969        if not hasattr(requester, "request_parameters"):
4970            return
4971
4972        request_parameters = requester.request_parameters
4973        if request_parameters and isinstance(request_parameters, Dict):
4974            for request_parameter_key in request_parameters.keys():
4975                request_parameter = request_parameters[request_parameter_key]
4976                if (
4977                    isinstance(request_parameter, Dict)
4978                    and request_parameter.get("type") == "QueryProperties"
4979                ):
4980                    request_parameters[request_parameter_key] = QueryPropertiesModel.parse_obj(
4981                        request_parameter
4982                    )
4983
4984    def _get_catalog_defined_cursor_field(
4985        self, stream_name: str, allow_catalog_defined_cursor_field: bool
4986    ) -> Optional[CursorField]:
4987        if not allow_catalog_defined_cursor_field:
4988            return None
4989
4990        configured_stream = self._stream_name_to_configured_stream.get(stream_name)
4991
4992        # Depending on the operation is being performed, there may not be a configured stream yet. In this
4993        # case we return None which will then use the default cursor field defined on the cursor model.
4994        # We also treat cursor_field: [""] (list with empty string) as no cursor field, since this can
4995        # occur when the platform serializes "no cursor configured" streams incorrectly.
4996        if (
4997            not configured_stream
4998            or not configured_stream.cursor_field
4999            or not configured_stream.cursor_field[0]
5000        ):
5001            return None
5002        elif len(configured_stream.cursor_field) > 1:
5003            raise ValueError(
5004                f"The `{stream_name}` stream does not support nested cursor_field. Please specify only a single cursor_field for the stream in the configured catalog."
5005            )
5006        else:
5007            return CursorField(
5008                cursor_field_key=configured_stream.cursor_field[0],
5009                supports_catalog_defined_cursor_field=allow_catalog_defined_cursor_field,
5010            )