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        return RecordExpander(
2510            expand_records_from_field=model.expand_records_from_field,
2511            config=config,
2512            parameters=model.parameters or {},
2513            remain_original_record=model.remain_original_record or False,
2514            on_no_records=OnNoRecords(model.on_no_records.value)
2515            if model.on_no_records
2516            else OnNoRecords.skip,
2517        )
2518
2519    @staticmethod
2520    def create_response_to_file_extractor(
2521        model: ResponseToFileExtractorModel,
2522        **kwargs: Any,
2523    ) -> ResponseToFileExtractor:
2524        return ResponseToFileExtractor(
2525            parameters=model.parameters or {},
2526            preserve_na_values=model.preserve_na_values or False,
2527        )
2528
2529    @staticmethod
2530    def create_exponential_backoff_strategy(
2531        model: ExponentialBackoffStrategyModel, config: Config
2532    ) -> ExponentialBackoffStrategy:
2533        ModelToComponentFactory._validate_jitter_range(model.jitter_range_in_seconds)
2534        return ExponentialBackoffStrategy(
2535            factor=model.factor or 5,
2536            jitter_range_in_seconds=model.jitter_range_in_seconds,
2537            parameters=model.parameters or {},
2538            config=config,
2539        )
2540
2541    @staticmethod
2542    def create_group_by_key(model: GroupByKeyMergeStrategyModel, config: Config) -> GroupByKey:
2543        return GroupByKey(model.key, config=config, parameters=model.parameters or {})
2544
2545    def create_http_requester(
2546        self,
2547        model: HttpRequesterModel,
2548        config: Config,
2549        decoder: Decoder = JsonDecoder(parameters={}),
2550        query_properties_key: Optional[str] = None,
2551        use_cache: Optional[bool] = None,
2552        *,
2553        name: str,
2554    ) -> HttpRequester:
2555        authenticator = (
2556            self._create_component_from_model(
2557                model=model.authenticator,
2558                config=config,
2559                url_base=model.url or model.url_base,
2560                name=name,
2561                decoder=decoder,
2562            )
2563            if model.authenticator
2564            else None
2565        )
2566        error_handler = (
2567            self._create_component_from_model(model=model.error_handler, config=config)
2568            if model.error_handler
2569            else DefaultErrorHandler(
2570                backoff_strategies=[],
2571                response_filters=[],
2572                config=config,
2573                parameters=model.parameters or {},
2574            )
2575        )
2576
2577        api_budget = self._api_budget
2578
2579        request_options_provider = InterpolatedRequestOptionsProvider(
2580            request_body=model.request_body,
2581            request_body_data=model.request_body_data,
2582            request_body_json=model.request_body_json,
2583            request_headers=model.request_headers,
2584            request_parameters=model.request_parameters,  # type: ignore  # QueryProperties have been removed in `create_simple_retriever`
2585            query_properties_key=query_properties_key,
2586            config=config,
2587            parameters=model.parameters or {},
2588        )
2589
2590        assert model.use_cache is not None  # for mypy
2591        assert model.http_method is not None  # for mypy
2592
2593        should_use_cache = (model.use_cache or bool(use_cache)) and not self._disable_cache
2594
2595        return HttpRequester(
2596            name=name,
2597            url=model.url,
2598            url_base=model.url_base,
2599            path=model.path,
2600            authenticator=authenticator,
2601            error_handler=error_handler,
2602            api_budget=api_budget,
2603            http_method=HttpMethod[model.http_method.value],
2604            request_options_provider=request_options_provider,
2605            config=config,
2606            disable_retries=self._disable_retries,
2607            parameters=model.parameters or {},
2608            message_repository=self._message_repository,
2609            use_cache=should_use_cache,
2610            decoder=decoder,
2611            stream_response=decoder.is_stream_response() if decoder else False,
2612        )
2613
2614    @staticmethod
2615    def create_http_response_filter(
2616        model: HttpResponseFilterModel, config: Config, **kwargs: Any
2617    ) -> HttpResponseFilter:
2618        if model.action:
2619            action = ResponseAction(model.action.value)
2620        else:
2621            action = None
2622
2623        failure_type = FailureType(model.failure_type.value) if model.failure_type else None
2624
2625        http_codes = (
2626            set(model.http_codes) if model.http_codes else set()
2627        )  # JSON schema notation has no set data type. The schema enforces an array of unique elements
2628
2629        return HttpResponseFilter(
2630            action=action,
2631            failure_type=failure_type,
2632            error_message=model.error_message or "",
2633            error_message_contains=model.error_message_contains or "",
2634            http_codes=http_codes,
2635            predicate=model.predicate or "",
2636            config=config,
2637            parameters=model.parameters or {},
2638        )
2639
2640    @staticmethod
2641    def create_inline_schema_loader(
2642        model: InlineSchemaLoaderModel, config: Config, **kwargs: Any
2643    ) -> InlineSchemaLoader:
2644        return InlineSchemaLoader(schema=model.schema_ or {}, parameters={})
2645
2646    def create_complex_field_type(
2647        self, model: ComplexFieldTypeModel, config: Config, **kwargs: Any
2648    ) -> ComplexFieldType:
2649        items = (
2650            self._create_component_from_model(model=model.items, config=config)
2651            if isinstance(model.items, ComplexFieldTypeModel)
2652            else model.items
2653        )
2654
2655        return ComplexFieldType(field_type=model.field_type, items=items)
2656
2657    def create_types_map(self, model: TypesMapModel, config: Config, **kwargs: Any) -> TypesMap:
2658        target_type = (
2659            self._create_component_from_model(model=model.target_type, config=config)
2660            if isinstance(model.target_type, ComplexFieldTypeModel)
2661            else model.target_type
2662        )
2663
2664        return TypesMap(
2665            target_type=target_type,
2666            current_type=model.current_type,
2667            condition=model.condition if model.condition is not None else "True",
2668        )
2669
2670    def create_schema_type_identifier(
2671        self, model: SchemaTypeIdentifierModel, config: Config, **kwargs: Any
2672    ) -> SchemaTypeIdentifier:
2673        types_mapping = []
2674        if model.types_mapping:
2675            types_mapping.extend(
2676                [
2677                    self._create_component_from_model(types_map, config=config)
2678                    for types_map in model.types_mapping
2679                ]
2680            )
2681        model_schema_pointer: List[Union[InterpolatedString, str]] = (
2682            [x for x in model.schema_pointer] if model.schema_pointer else []
2683        )
2684        model_key_pointer: List[Union[InterpolatedString, str]] = [x for x in model.key_pointer]
2685        model_type_pointer: Optional[List[Union[InterpolatedString, str]]] = (
2686            [x for x in model.type_pointer] if model.type_pointer else None
2687        )
2688
2689        return SchemaTypeIdentifier(
2690            schema_pointer=model_schema_pointer,
2691            key_pointer=model_key_pointer,
2692            type_pointer=model_type_pointer,
2693            types_mapping=types_mapping,
2694            parameters=model.parameters or {},
2695        )
2696
2697    def create_dynamic_schema_loader(
2698        self, model: DynamicSchemaLoaderModel, config: Config, **kwargs: Any
2699    ) -> DynamicSchemaLoader:
2700        schema_transformations = []
2701        if model.schema_transformations:
2702            for transformation_model in model.schema_transformations:
2703                schema_transformations.append(
2704                    self._create_component_from_model(model=transformation_model, config=config)
2705                )
2706        name = "dynamic_properties"
2707        retriever = self._create_component_from_model(
2708            model=model.retriever,
2709            config=config,
2710            name=name,
2711            primary_key=None,
2712            partition_router=self._build_stream_slicer_from_partition_router(
2713                model.retriever, config
2714            ),
2715            transformations=[],
2716            use_cache=True,
2717            log_formatter=(
2718                lambda response: format_http_message(
2719                    response,
2720                    f"Schema loader '{name}' request",
2721                    f"Request performed in order to extract schema.",
2722                    name,
2723                    is_auxiliary=True,
2724                )
2725            ),
2726        )
2727        schema_type_identifier = self._create_component_from_model(
2728            model.schema_type_identifier, config=config, parameters=model.parameters or {}
2729        )
2730        schema_filter = (
2731            self._create_component_from_model(
2732                model.schema_filter, config=config, parameters=model.parameters or {}
2733            )
2734            if model.schema_filter is not None
2735            else None
2736        )
2737
2738        return DynamicSchemaLoader(
2739            retriever=retriever,
2740            config=config,
2741            schema_transformations=schema_transformations,
2742            schema_filter=schema_filter,
2743            schema_type_identifier=schema_type_identifier,
2744            parameters=model.parameters or {},
2745        )
2746
2747    @staticmethod
2748    def create_json_decoder(model: JsonDecoderModel, config: Config, **kwargs: Any) -> Decoder:
2749        return JsonDecoder(parameters={})
2750
2751    def create_csv_decoder(self, model: CsvDecoderModel, config: Config, **kwargs: Any) -> Decoder:
2752        return CompositeRawDecoder(
2753            parser=ModelToComponentFactory._get_parser(model, config),
2754            stream_response=False if self._emit_connector_builder_messages else True,
2755        )
2756
2757    def create_jsonl_decoder(
2758        self, model: JsonlDecoderModel, config: Config, **kwargs: Any
2759    ) -> Decoder:
2760        return CompositeRawDecoder(
2761            parser=ModelToComponentFactory._get_parser(model, config),
2762            stream_response=False if self._emit_connector_builder_messages else True,
2763        )
2764
2765    def create_json_items_decoder(
2766        self, model: JsonItemsDecoderModel, config: Config, **kwargs: Any
2767    ) -> Decoder:
2768        return CompositeRawDecoder(
2769            parser=ModelToComponentFactory._get_parser(model, config),
2770            stream_response=False if self._emit_connector_builder_messages else True,
2771        )
2772
2773    def create_gzip_decoder(
2774        self, model: GzipDecoderModel, config: Config, **kwargs: Any
2775    ) -> Decoder:
2776        _compressed_response_types = {
2777            "gzip",
2778            "x-gzip",
2779            "gzip, deflate",
2780            "x-gzip, deflate",
2781            "application/zip",
2782            "application/gzip",
2783            "application/x-gzip",
2784            "application/x-zip-compressed",
2785        }
2786
2787        gzip_parser: GzipParser = ModelToComponentFactory._get_parser(model, config)  # type: ignore  # based on the model, we know this will be a GzipParser
2788
2789        if self._emit_connector_builder_messages:
2790            return CompositeRawDecoder(gzip_parser, False)
2791
2792        transport_gzip_parser = GzipParser(inner_parser=gzip_parser)
2793        return CompositeRawDecoder.by_headers(
2794            [
2795                ({"Content-Encoding"}, {"gzip"}, transport_gzip_parser),
2796                ({"Content-Type"}, _compressed_response_types, gzip_parser),
2797            ],
2798            stream_response=True,
2799            fallback_parser=gzip_parser,
2800        )
2801
2802    @staticmethod
2803    def create_iterable_decoder(
2804        model: IterableDecoderModel, config: Config, **kwargs: Any
2805    ) -> IterableDecoder:
2806        return IterableDecoder(parameters={})
2807
2808    @staticmethod
2809    def create_xml_decoder(model: XmlDecoderModel, config: Config, **kwargs: Any) -> XmlDecoder:
2810        return XmlDecoder(parameters={})
2811
2812    def create_zipfile_decoder(
2813        self, model: ZipfileDecoderModel, config: Config, **kwargs: Any
2814    ) -> ZipfileDecoder:
2815        return ZipfileDecoder(parser=ModelToComponentFactory._get_parser(model.decoder, config))
2816
2817    @staticmethod
2818    def _get_parser(model: BaseModel, config: Config) -> Parser:
2819        if isinstance(model, JsonDecoderModel):
2820            # Note that the logic is a bit different from the JsonDecoder as there is some legacy that is maintained to return {} on error cases
2821            return JsonParser()
2822        elif isinstance(model, JsonItemsDecoderModel):
2823            return JsonItemsParser(
2824                items_path=model.items_path,
2825                encoding=model.encoding,
2826            )
2827        elif isinstance(model, JsonlDecoderModel):
2828            return JsonLineParser()
2829        elif isinstance(model, CsvDecoderModel):
2830            return CsvParser(
2831                encoding=model.encoding,
2832                delimiter=model.delimiter,
2833                set_values_to_none=model.set_values_to_none,
2834            )
2835        elif isinstance(model, GzipDecoderModel):
2836            return GzipParser(
2837                inner_parser=ModelToComponentFactory._get_parser(model.decoder, config)
2838            )
2839        elif isinstance(
2840            model, (CustomDecoderModel, IterableDecoderModel, XmlDecoderModel, ZipfileDecoderModel)
2841        ):
2842            raise ValueError(f"Decoder type {model} does not have parser associated to it")
2843
2844        raise ValueError(f"Unknown decoder type {model}")
2845
2846    @staticmethod
2847    def create_json_file_schema_loader(
2848        model: JsonFileSchemaLoaderModel, config: Config, **kwargs: Any
2849    ) -> JsonFileSchemaLoader:
2850        return JsonFileSchemaLoader(
2851            file_path=model.file_path or "", config=config, parameters=model.parameters or {}
2852        )
2853
2854    def create_jwt_authenticator(
2855        self, model: JwtAuthenticatorModel, config: Config, **kwargs: Any
2856    ) -> JwtAuthenticator:
2857        jwt_headers = model.jwt_headers or JwtHeadersModel(kid=None, typ="JWT", cty=None)
2858        jwt_payload = model.jwt_payload or JwtPayloadModel(iss=None, sub=None, aud=None)
2859        request_option = (
2860            self._create_component_from_model(model.request_option, config)
2861            if model.request_option
2862            else None
2863        )
2864        return JwtAuthenticator(
2865            config=config,
2866            parameters=model.parameters or {},
2867            algorithm=JwtAlgorithm(model.algorithm.value),
2868            secret_key=model.secret_key,
2869            base64_encode_secret_key=model.base64_encode_secret_key,
2870            token_duration=model.token_duration,
2871            header_prefix=model.header_prefix,
2872            kid=jwt_headers.kid,
2873            typ=jwt_headers.typ,
2874            cty=jwt_headers.cty,
2875            iss=jwt_payload.iss,
2876            sub=jwt_payload.sub,
2877            aud=jwt_payload.aud,
2878            additional_jwt_headers=model.additional_jwt_headers,
2879            additional_jwt_payload=model.additional_jwt_payload,
2880            passphrase=model.passphrase,
2881            request_option=request_option,
2882        )
2883
2884    def create_list_partition_router(
2885        self, model: ListPartitionRouterModel, config: Config, **kwargs: Any
2886    ) -> ListPartitionRouter:
2887        request_option = (
2888            self._create_component_from_model(model.request_option, config)
2889            if model.request_option
2890            else None
2891        )
2892        return ListPartitionRouter(
2893            cursor_field=model.cursor_field,
2894            request_option=request_option,
2895            values=model.values,
2896            config=config,
2897            parameters=model.parameters or {},
2898        )
2899
2900    @staticmethod
2901    def create_min_max_datetime(
2902        model: MinMaxDatetimeModel, config: Config, **kwargs: Any
2903    ) -> MinMaxDatetime:
2904        return MinMaxDatetime(
2905            datetime=model.datetime,
2906            datetime_format=model.datetime_format or "",
2907            max_datetime=model.max_datetime or "",
2908            min_datetime=model.min_datetime or "",
2909            parameters=model.parameters or {},
2910        )
2911
2912    @staticmethod
2913    def create_no_auth(model: NoAuthModel, config: Config, **kwargs: Any) -> NoAuth:
2914        return NoAuth(parameters=model.parameters or {})
2915
2916    @staticmethod
2917    def create_no_pagination(
2918        model: NoPaginationModel, config: Config, **kwargs: Any
2919    ) -> NoPagination:
2920        return NoPagination(parameters={})
2921
2922    def create_oauth_authenticator(
2923        self, model: OAuthAuthenticatorModel, config: Config, **kwargs: Any
2924    ) -> DeclarativeOauth2Authenticator:
2925        profile_assertion = (
2926            self._create_component_from_model(model.profile_assertion, config=config)
2927            if model.profile_assertion
2928            else None
2929        )
2930
2931        refresh_token_error_status_codes, refresh_token_error_key, refresh_token_error_values = (
2932            self._get_refresh_token_error_information(model)
2933        )
2934        if model.refresh_token_updater:
2935            # ignore type error because fixing it would have a lot of dependencies, revisit later
2936            return DeclarativeSingleUseRefreshTokenOauth2Authenticator(  # type: ignore
2937                config,
2938                InterpolatedString.create(
2939                    model.token_refresh_endpoint,  # type: ignore
2940                    parameters=model.parameters or {},
2941                ).eval(config),
2942                access_token_name=InterpolatedString.create(
2943                    model.access_token_name or "access_token", parameters=model.parameters or {}
2944                ).eval(config),
2945                refresh_token_name=model.refresh_token_updater.refresh_token_name,
2946                expires_in_name=InterpolatedString.create(
2947                    model.expires_in_name or "expires_in", parameters=model.parameters or {}
2948                ).eval(config),
2949                client_id_name=InterpolatedString.create(
2950                    model.client_id_name or "client_id", parameters=model.parameters or {}
2951                ).eval(config),
2952                client_id=InterpolatedString.create(
2953                    model.client_id, parameters=model.parameters or {}
2954                ).eval(config)
2955                if model.client_id
2956                else model.client_id,
2957                client_secret_name=InterpolatedString.create(
2958                    model.client_secret_name or "client_secret", parameters=model.parameters or {}
2959                ).eval(config),
2960                client_secret=InterpolatedString.create(
2961                    model.client_secret, parameters=model.parameters or {}
2962                ).eval(config)
2963                if model.client_secret
2964                else model.client_secret,
2965                access_token_config_path=model.refresh_token_updater.access_token_config_path,
2966                refresh_token_config_path=model.refresh_token_updater.refresh_token_config_path,
2967                token_expiry_date_config_path=model.refresh_token_updater.token_expiry_date_config_path,
2968                grant_type_name=InterpolatedString.create(
2969                    model.grant_type_name or "grant_type", parameters=model.parameters or {}
2970                ).eval(config),
2971                grant_type=InterpolatedString.create(
2972                    model.grant_type or "refresh_token", parameters=model.parameters or {}
2973                ).eval(config),
2974                refresh_request_body=InterpolatedMapping(
2975                    model.refresh_request_body or {}, parameters=model.parameters or {}
2976                ).eval(config),
2977                refresh_request_headers=InterpolatedMapping(
2978                    model.refresh_request_headers or {}, parameters=model.parameters or {}
2979                ).eval(config),
2980                send_refresh_request_as_query_params=bool(
2981                    model.send_refresh_request_as_query_params
2982                ),
2983                scopes=model.scopes,
2984                token_expiry_date_format=model.token_expiry_date_format,
2985                token_expiry_is_time_of_expiration=bool(model.token_expiry_date_format),
2986                message_repository=self._message_repository,
2987                refresh_token_error_status_codes=refresh_token_error_status_codes,
2988                refresh_token_error_key=refresh_token_error_key,
2989                refresh_token_error_values=refresh_token_error_values,
2990            )
2991        # ignore type error because fixing it would have a lot of dependencies, revisit later
2992        return DeclarativeOauth2Authenticator(  # type: ignore
2993            access_token_name=model.access_token_name or "access_token",
2994            access_token_value=model.access_token_value,
2995            client_id_name=model.client_id_name or "client_id",
2996            client_id=model.client_id,
2997            client_secret_name=model.client_secret_name or "client_secret",
2998            client_secret=model.client_secret,
2999            expires_in_name=model.expires_in_name or "expires_in",
3000            grant_type_name=model.grant_type_name or "grant_type",
3001            grant_type=model.grant_type or "refresh_token",
3002            refresh_request_body=model.refresh_request_body,
3003            refresh_request_headers=model.refresh_request_headers,
3004            send_refresh_request_as_query_params=bool(model.send_refresh_request_as_query_params),
3005            refresh_token_name=model.refresh_token_name or "refresh_token",
3006            refresh_token=model.refresh_token,
3007            scopes=model.scopes,
3008            token_expiry_date=model.token_expiry_date,
3009            token_expiry_date_format=model.token_expiry_date_format,
3010            token_expiry_is_time_of_expiration=bool(model.token_expiry_date_format),
3011            token_refresh_endpoint=model.token_refresh_endpoint,
3012            config=config,
3013            parameters=model.parameters or {},
3014            message_repository=self._message_repository,
3015            profile_assertion=profile_assertion,
3016            use_profile_assertion=model.use_profile_assertion,
3017            refresh_token_error_status_codes=refresh_token_error_status_codes,
3018            refresh_token_error_key=refresh_token_error_key,
3019            refresh_token_error_values=refresh_token_error_values,
3020        )
3021
3022    @staticmethod
3023    def _get_refresh_token_error_information(
3024        model: OAuthAuthenticatorModel,
3025    ) -> Tuple[Tuple[int, ...], str, Tuple[str, ...]]:
3026        """
3027        In a previous version of the CDK, the auth error as config_error was only done if a refresh token updater was
3028        defined. As a transition, we added those fields on the OAuthAuthenticatorModel. This method ensures that the
3029        information is defined only once and return the right fields.
3030        """
3031        refresh_token_updater = model.refresh_token_updater
3032        is_defined_on_refresh_token_updated = refresh_token_updater and (
3033            refresh_token_updater.refresh_token_error_status_codes
3034            or refresh_token_updater.refresh_token_error_key
3035            or refresh_token_updater.refresh_token_error_values
3036        )
3037        is_defined_on_oauth_authenticator = (
3038            model.refresh_token_error_status_codes
3039            or model.refresh_token_error_key
3040            or model.refresh_token_error_values
3041        )
3042        if is_defined_on_refresh_token_updated and is_defined_on_oauth_authenticator:
3043            raise ValueError(
3044                "refresh_token_error should either be defined on the OAuthAuthenticatorModel or the RefreshTokenUpdaterModel, not both"
3045            )
3046
3047        if is_defined_on_refresh_token_updated:
3048            not_optional_refresh_token_updater: RefreshTokenUpdaterModel = refresh_token_updater  # type: ignore  # we know from the condition that this is not None
3049            return (
3050                tuple(not_optional_refresh_token_updater.refresh_token_error_status_codes)
3051                if not_optional_refresh_token_updater.refresh_token_error_status_codes
3052                else (),
3053                not_optional_refresh_token_updater.refresh_token_error_key or "",
3054                tuple(not_optional_refresh_token_updater.refresh_token_error_values)
3055                if not_optional_refresh_token_updater.refresh_token_error_values
3056                else (),
3057            )
3058        elif is_defined_on_oauth_authenticator:
3059            return (
3060                tuple(model.refresh_token_error_status_codes)
3061                if model.refresh_token_error_status_codes
3062                else (),
3063                model.refresh_token_error_key or "",
3064                tuple(model.refresh_token_error_values) if model.refresh_token_error_values else (),
3065            )
3066
3067        # returning default values we think cover most cases
3068        return (400,), "error", ("invalid_grant", "invalid_permissions")
3069
3070    def create_offset_increment(
3071        self,
3072        model: OffsetIncrementModel,
3073        config: Config,
3074        decoder: Decoder,
3075        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
3076        **kwargs: Any,
3077    ) -> OffsetIncrement:
3078        if isinstance(decoder, PaginationDecoderDecorator):
3079            inner_decoder = decoder.decoder
3080        else:
3081            inner_decoder = decoder
3082            decoder = PaginationDecoderDecorator(decoder=decoder)
3083
3084        if self._is_supported_decoder_for_pagination(inner_decoder):
3085            decoder_to_use = decoder
3086        else:
3087            raise ValueError(
3088                self._UNSUPPORTED_DECODER_ERROR.format(decoder_type=type(inner_decoder))
3089            )
3090
3091        # Ideally we would instantiate the runtime extractor from highest most level (in this case the SimpleRetriever)
3092        # so that it can be shared by OffSetIncrement and RecordSelector. However, due to how we instantiate the
3093        # decoder with various decorators here, but not in create_record_selector, it is simpler to retain existing
3094        # behavior by having two separate extractors with identical behavior since they use the same extractor model.
3095        # When we have more time to investigate we can look into reusing the same component.
3096        extractor = (
3097            self._create_component_from_model(
3098                model=extractor_model, config=config, decoder=decoder_to_use
3099            )
3100            if extractor_model
3101            else None
3102        )
3103
3104        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
3105        # If page_size is a string that represents an integer (not an interpolation), convert it back.
3106        page_size = model.page_size
3107        if isinstance(page_size, str) and page_size.isdigit():
3108            page_size = int(page_size)
3109
3110        return OffsetIncrement(
3111            page_size=page_size,
3112            config=config,
3113            decoder=decoder_to_use,
3114            extractor=extractor,
3115            inject_on_first_request=model.inject_on_first_request or False,
3116            parameters=model.parameters or {},
3117        )
3118
3119    def create_page_increment(
3120        self,
3121        model: PageIncrementModel,
3122        config: Config,
3123        decoder: Optional[Decoder] = None,
3124        extractor_model: Optional[Union[CustomRecordExtractorModel, DpathExtractorModel]] = None,
3125        **kwargs: Any,
3126    ) -> PageIncrement:
3127        # Like OffsetIncrement, we instantiate a separate extractor with identical behavior to the
3128        # RecordSelector's so the strategy can count the raw records in the response. This ensures
3129        # pagination is driven by the API's page size, not the post-filter record count.
3130        extractor = (
3131            self._create_component_from_model(model=extractor_model, config=config, decoder=decoder)
3132            if extractor_model
3133            else None
3134        )
3135
3136        # Pydantic v1 Union type coercion can convert int to string depending on Union order.
3137        # If page_size is a string that represents an integer (not an interpolation), convert it back.
3138        page_size = model.page_size
3139        if isinstance(page_size, str) and page_size.isdigit():
3140            page_size = int(page_size)
3141
3142        return PageIncrement(
3143            page_size=page_size,
3144            config=config,
3145            start_from_page=model.start_from_page or 0,
3146            inject_on_first_request=model.inject_on_first_request or False,
3147            extractor=extractor,
3148            parameters=model.parameters or {},
3149        )
3150
3151    def create_parent_stream_config(
3152        self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any
3153    ) -> ParentStreamConfig:
3154        declarative_stream = self._create_component_from_model(
3155            model.stream,
3156            config=config,
3157            is_parent=True,
3158            **kwargs,
3159        )
3160        request_option = (
3161            self._create_component_from_model(model.request_option, config=config)
3162            if model.request_option
3163            else None
3164        )
3165
3166        if model.lazy_read_pointer and any("*" in pointer for pointer in model.lazy_read_pointer):
3167            raise ValueError(
3168                "The '*' wildcard in 'lazy_read_pointer' is not supported — only direct paths are allowed."
3169            )
3170
3171        model_lazy_read_pointer: List[Union[InterpolatedString, str]] = (
3172            [x for x in model.lazy_read_pointer] if model.lazy_read_pointer else []
3173        )
3174
3175        return ParentStreamConfig(
3176            parent_key=model.parent_key,
3177            request_option=request_option,
3178            stream=declarative_stream,
3179            partition_field=model.partition_field,
3180            config=config,
3181            incremental_dependency=model.incremental_dependency or False,
3182            parameters=model.parameters or {},
3183            extra_fields=model.extra_fields,
3184            lazy_read_pointer=model_lazy_read_pointer,
3185        )
3186
3187    def create_properties_from_endpoint(
3188        self, model: PropertiesFromEndpointModel, config: Config, **kwargs: Any
3189    ) -> PropertiesFromEndpoint:
3190        retriever = self._create_component_from_model(
3191            model=model.retriever,
3192            config=config,
3193            name="dynamic_properties",
3194            primary_key=None,
3195            stream_slicer=None,
3196            transformations=[],
3197            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
3198        )
3199        return PropertiesFromEndpoint(
3200            property_field_path=model.property_field_path,
3201            retriever=retriever,
3202            config=config,
3203            parameters=model.parameters or {},
3204        )
3205
3206    def create_property_chunking(
3207        self, model: PropertyChunkingModel, config: Config, **kwargs: Any
3208    ) -> PropertyChunking:
3209        record_merge_strategy = (
3210            self._create_component_from_model(
3211                model=model.record_merge_strategy, config=config, **kwargs
3212            )
3213            if model.record_merge_strategy
3214            else None
3215        )
3216
3217        property_limit_type: PropertyLimitType
3218        match model.property_limit_type:
3219            case PropertyLimitTypeModel.property_count:
3220                property_limit_type = PropertyLimitType.property_count
3221            case PropertyLimitTypeModel.characters:
3222                property_limit_type = PropertyLimitType.characters
3223            case _:
3224                raise ValueError(f"Invalid PropertyLimitType {property_limit_type}")
3225
3226        return PropertyChunking(
3227            property_limit_type=property_limit_type,
3228            property_limit=model.property_limit,
3229            record_merge_strategy=record_merge_strategy,
3230            config=config,
3231            parameters=model.parameters or {},
3232        )
3233
3234    def create_query_properties(
3235        self, model: QueryPropertiesModel, config: Config, *, stream_name: str, **kwargs: Any
3236    ) -> QueryProperties:
3237        if isinstance(model.property_list, list):
3238            property_list = model.property_list
3239        else:
3240            property_list = self._create_component_from_model(
3241                model=model.property_list, config=config, **kwargs
3242            )
3243
3244        property_chunking = (
3245            self._create_component_from_model(
3246                model=model.property_chunking, config=config, **kwargs
3247            )
3248            if model.property_chunking
3249            else None
3250        )
3251
3252        property_selector = (
3253            self._create_component_from_model(
3254                model=model.property_selector, config=config, stream_name=stream_name, **kwargs
3255            )
3256            if model.property_selector
3257            else None
3258        )
3259
3260        return QueryProperties(
3261            property_list=property_list,
3262            always_include_properties=model.always_include_properties,
3263            property_chunking=property_chunking,
3264            property_selector=property_selector,
3265            config=config,
3266            parameters=model.parameters or {},
3267        )
3268
3269    def create_json_schema_property_selector(
3270        self,
3271        model: JsonSchemaPropertySelectorModel,
3272        config: Config,
3273        *,
3274        stream_name: str,
3275        **kwargs: Any,
3276    ) -> JsonSchemaPropertySelector:
3277        configured_stream = self._stream_name_to_configured_stream.get(stream_name)
3278
3279        transformations = []
3280        if model.transformations:
3281            for transformation_model in model.transformations:
3282                transformations.append(
3283                    self._create_component_from_model(model=transformation_model, config=config)
3284                )
3285
3286        return JsonSchemaPropertySelector(
3287            configured_stream=configured_stream,
3288            properties_transformations=transformations,
3289            config=config,
3290            parameters=model.parameters or {},
3291        )
3292
3293    @staticmethod
3294    def create_record_filter(
3295        model: RecordFilterModel, config: Config, **kwargs: Any
3296    ) -> RecordFilter:
3297        return RecordFilter(
3298            condition=model.condition or "", config=config, parameters=model.parameters or {}
3299        )
3300
3301    @staticmethod
3302    def create_request_path(model: RequestPathModel, config: Config, **kwargs: Any) -> RequestPath:
3303        return RequestPath(parameters={})
3304
3305    @staticmethod
3306    def create_request_option(
3307        model: RequestOptionModel, config: Config, **kwargs: Any
3308    ) -> RequestOption:
3309        inject_into = RequestOptionType(model.inject_into.value)
3310        field_path: Optional[List[Union[InterpolatedString, str]]] = (
3311            [
3312                InterpolatedString.create(segment, parameters=kwargs.get("parameters", {}))
3313                for segment in model.field_path
3314            ]
3315            if model.field_path
3316            else None
3317        )
3318        field_name = (
3319            InterpolatedString.create(model.field_name, parameters=kwargs.get("parameters", {}))
3320            if model.field_name
3321            else None
3322        )
3323        return RequestOption(
3324            field_name=field_name,
3325            field_path=field_path,
3326            inject_into=inject_into,
3327            parameters=kwargs.get("parameters", {}),
3328        )
3329
3330    def create_record_selector(
3331        self,
3332        model: RecordSelectorModel,
3333        config: Config,
3334        *,
3335        name: str,
3336        transformations: List[RecordTransformation] | None = None,
3337        decoder: Decoder | None = None,
3338        client_side_incremental_sync_cursor: Optional[Cursor] = None,
3339        is_client_side_incremental_sync: bool = False,
3340        file_uploader: Optional[DefaultFileUploader] = None,
3341        **kwargs: Any,
3342    ) -> RecordSelector:
3343        extractor = self._create_component_from_model(
3344            model=model.extractor, decoder=decoder, config=config
3345        )
3346        record_filter = (
3347            self._create_component_from_model(model.record_filter, config=config)
3348            if model.record_filter
3349            else None
3350        )
3351
3352        # A client-side incremental stream transforms before filtering by default. That default belongs to the flag,
3353        # not to the component that ends up doing the cursor comparison: a data feed does it in the retriever and
3354        # receives no cursor here, but its `record_filter` condition must keep running after the transformations.
3355        default_transform_before_filtering = bool(
3356            client_side_incremental_sync_cursor or is_client_side_incremental_sync
3357        )
3358        transform_before_filtering = (
3359            default_transform_before_filtering
3360            if model.transform_before_filtering is None
3361            else model.transform_before_filtering
3362        )
3363        if client_side_incremental_sync_cursor:
3364            record_filter = ClientSideIncrementalRecordFilterDecorator(
3365                config=config,
3366                parameters=model.parameters,
3367                condition=model.record_filter.condition
3368                if (model.record_filter and hasattr(model.record_filter, "condition"))
3369                else None,
3370                cursor=client_side_incremental_sync_cursor,
3371            )
3372
3373        if model.schema_normalization is None:
3374            # default to no schema normalization if not set
3375            model.schema_normalization = SchemaNormalizationModel.None_
3376
3377        schema_normalization = (
3378            TypeTransformer(SCHEMA_TRANSFORMER_TYPE_MAPPING[model.schema_normalization])
3379            if isinstance(model.schema_normalization, SchemaNormalizationModel)
3380            else self._create_component_from_model(model.schema_normalization, config=config)  # type: ignore[arg-type] # custom normalization model expected here
3381        )
3382
3383        return RecordSelector(
3384            extractor=extractor,
3385            name=name,
3386            config=config,
3387            record_filter=record_filter,
3388            transformations=transformations or [],
3389            file_uploader=file_uploader,
3390            schema_normalization=schema_normalization,
3391            parameters=model.parameters or {},
3392            transform_before_filtering=transform_before_filtering,
3393        )
3394
3395    @staticmethod
3396    def create_remove_fields(
3397        model: RemoveFieldsModel, config: Config, **kwargs: Any
3398    ) -> RemoveFields:
3399        return RemoveFields(
3400            field_pointers=model.field_pointers, condition=model.condition or "", parameters={}
3401        )
3402
3403    def create_selective_authenticator(
3404        self, model: SelectiveAuthenticatorModel, config: Config, **kwargs: Any
3405    ) -> DeclarativeAuthenticator:
3406        authenticators = {
3407            name: self._create_component_from_model(model=auth, config=config)
3408            for name, auth in model.authenticators.items()
3409        }
3410        # SelectiveAuthenticator will return instance of DeclarativeAuthenticator or raise ValueError error
3411        return SelectiveAuthenticator(  # type: ignore[abstract]
3412            config=config,
3413            authenticators=authenticators,
3414            authenticator_selection_path=model.authenticator_selection_path,
3415            **kwargs,
3416        )
3417
3418    @staticmethod
3419    def create_legacy_session_token_authenticator(
3420        model: LegacySessionTokenAuthenticatorModel, config: Config, *, url_base: str, **kwargs: Any
3421    ) -> LegacySessionTokenAuthenticator:
3422        return LegacySessionTokenAuthenticator(
3423            api_url=url_base,
3424            header=model.header,
3425            login_url=model.login_url,
3426            password=model.password or "",
3427            session_token=model.session_token or "",
3428            session_token_response_key=model.session_token_response_key or "",
3429            username=model.username or "",
3430            validate_session_url=model.validate_session_url,
3431            config=config,
3432            parameters=model.parameters or {},
3433        )
3434
3435    def create_simple_retriever(
3436        self,
3437        model: SimpleRetrieverModel,
3438        config: Config,
3439        *,
3440        name: str,
3441        primary_key: Optional[Union[str, List[str], List[List[str]]]],
3442        request_options_provider: Optional[RequestOptionsProvider] = None,
3443        cursor: Optional[Cursor] = None,
3444        has_stop_condition_cursor: bool = False,
3445        is_client_side_incremental_sync: bool = False,
3446        transformations: List[RecordTransformation],
3447        file_uploader: Optional[DefaultFileUploader] = None,
3448        incremental_sync: Optional[
3449            Union[IncrementingCountCursorModel, DatetimeBasedCursorModel]
3450        ] = None,
3451        use_cache: Optional[bool] = None,
3452        log_formatter: Optional[Callable[[Response], Any]] = None,
3453        partition_router: Optional[PartitionRouter] = None,
3454        **kwargs: Any,
3455    ) -> SimpleRetriever:
3456        def _get_url(req: Requester) -> str:
3457            """
3458            Closure to get the URL from the requester. This is used to get the URL in the case of a lazy retriever.
3459            This is needed because the URL is not set until the requester is created.
3460            """
3461
3462            _url: str = (
3463                model.requester.url
3464                if hasattr(model.requester, "url") and model.requester.url is not None
3465                else req.get_url(stream_state=None, stream_slice=None, next_page_token=None)
3466            )
3467            _url_base: str = (
3468                model.requester.url_base
3469                if hasattr(model.requester, "url_base") and model.requester.url_base is not None
3470                else req.get_url_base(stream_state=None, stream_slice=None, next_page_token=None)
3471            )
3472
3473            return _url or _url_base
3474
3475        if cursor is None:
3476            cursor = FinalStateCursor(name, None, self._message_repository)
3477
3478        # A data feed drops the records the cursor considers already synced in the retriever, which sits downstream of
3479        # the paginator. Letting the record selector drop them as well would be redundant and would hide them from the
3480        # pagination stop condition, so a data feed never delegates that filtering to the record selector, whether
3481        # `is_client_side_incremental` is set or not. The `condition` from `record_filter` is intentionally left out of
3482        # the post-pagination filter and stays in the record selector, which preserves the existing behaviour: the
3483        # selector runs inside the page loop, so the records the condition rejects never reach the paginator's
3484        # accounting. Moving it downstream would start counting them.
3485        post_pagination_filter = (
3486            ClientSideIncrementalRecordFilterDecorator(
3487                config=config,
3488                parameters=model.parameters or {},
3489                condition=None,
3490                cursor=cursor,
3491            )
3492            if has_stop_condition_cursor
3493            else None
3494        )
3495        client_side_incremental_cursor = (
3496            cursor if is_client_side_incremental_sync and not post_pagination_filter else None
3497        )
3498        if post_pagination_filter and is_client_side_incremental_sync:
3499            LOGGER.warning(
3500                f"Stream {name}: `is_client_side_incremental` adds no record filtering when `is_data_feed` is set, "
3501                "as a data feed already filters on the cursor value. It still makes the record selector apply the "
3502                "transformations before the `record_filter` condition."
3503            )
3504
3505        decoder = (
3506            self._create_component_from_model(model=model.decoder, config=config)
3507            if model.decoder
3508            else JsonDecoder(parameters={})
3509        )
3510        record_selector = self._create_component_from_model(
3511            model=model.record_selector,
3512            name=name,
3513            config=config,
3514            decoder=decoder,
3515            transformations=transformations,
3516            client_side_incremental_sync_cursor=client_side_incremental_cursor,
3517            is_client_side_incremental_sync=is_client_side_incremental_sync,
3518            file_uploader=file_uploader,
3519        )
3520
3521        query_properties: Optional[QueryProperties] = None
3522        query_properties_key: Optional[str] = None
3523        self._ensure_query_properties_to_model(model.requester)
3524        if self._has_query_properties_in_request_parameters(model.requester):
3525            # It is better to be explicit about an error if PropertiesFromEndpoint is defined in multiple
3526            # places instead of default to request_parameters which isn't clearly documented
3527            if (
3528                hasattr(model.requester, "fetch_properties_from_endpoint")
3529                and model.requester.fetch_properties_from_endpoint
3530            ):
3531                raise ValueError(
3532                    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"
3533                )
3534
3535            query_properties_definitions = []
3536            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()
3537                if isinstance(request_parameter, QueryPropertiesModel):
3538                    query_properties_key = key
3539                    query_properties_definitions.append(request_parameter)
3540
3541            if len(query_properties_definitions) > 1:
3542                raise ValueError(
3543                    f"request_parameters only supports defining one QueryProperties field, but found {len(query_properties_definitions)} usages"
3544                )
3545
3546            if len(query_properties_definitions) == 1:
3547                query_properties = self._create_component_from_model(
3548                    model=query_properties_definitions[0], stream_name=name, config=config
3549                )
3550
3551            # Removes QueryProperties components from the interpolated mappings because it has been designed
3552            # to be used by the SimpleRetriever and will be resolved from the provider from the slice directly
3553            # instead of through jinja interpolation
3554            if hasattr(model.requester, "request_parameters") and isinstance(
3555                model.requester.request_parameters, Mapping
3556            ):
3557                model.requester.request_parameters = self._remove_query_properties(
3558                    model.requester.request_parameters
3559                )
3560        elif (
3561            hasattr(model.requester, "fetch_properties_from_endpoint")
3562            and model.requester.fetch_properties_from_endpoint
3563        ):
3564            # todo: Deprecate this condition once dependent connectors migrate to query_properties
3565            query_properties_definition = QueryPropertiesModel(
3566                type="QueryProperties",
3567                property_list=model.requester.fetch_properties_from_endpoint,
3568                always_include_properties=None,
3569                property_chunking=None,
3570            )  # type: ignore # $parameters has a default value
3571
3572            query_properties = self.create_query_properties(
3573                model=query_properties_definition,
3574                stream_name=name,
3575                config=config,
3576            )
3577        elif hasattr(model.requester, "query_properties") and model.requester.query_properties:
3578            query_properties = self.create_query_properties(
3579                model=model.requester.query_properties,
3580                stream_name=name,
3581                config=config,
3582            )
3583
3584        requester = self._create_component_from_model(
3585            model=model.requester,
3586            decoder=decoder,
3587            name=name,
3588            query_properties_key=query_properties_key,
3589            use_cache=use_cache,
3590            config=config,
3591        )
3592
3593        if not request_options_provider:
3594            request_options_provider = DefaultRequestOptionsProvider(parameters={})
3595        if isinstance(request_options_provider, DefaultRequestOptionsProvider) and isinstance(
3596            partition_router, PartitionRouter
3597        ):
3598            request_options_provider = partition_router
3599
3600        paginator = (
3601            self._create_component_from_model(
3602                model=model.paginator,
3603                config=config,
3604                url_base=_get_url(requester),
3605                extractor_model=model.record_selector.extractor,
3606                decoder=decoder,
3607                cursor_used_for_stop_condition=cursor if has_stop_condition_cursor else None,
3608            )
3609            if model.paginator
3610            else NoPagination(parameters={})
3611        )
3612
3613        ignore_stream_slicer_parameters_on_paginated_requests = (
3614            model.ignore_stream_slicer_parameters_on_paginated_requests or False
3615        )
3616
3617        if (
3618            model.partition_router
3619            and isinstance(model.partition_router, SubstreamPartitionRouterModel)
3620            and not bool(self._connector_state_manager.get_stream_state(name, None))
3621            and any(
3622                parent_stream_config.lazy_read_pointer
3623                for parent_stream_config in model.partition_router.parent_stream_configs
3624            )
3625        ):
3626            if incremental_sync:
3627                if incremental_sync.type != "DatetimeBasedCursor":
3628                    raise ValueError(
3629                        f"LazySimpleRetriever only supports DatetimeBasedCursor. Found: {incremental_sync.type}."
3630                    )
3631
3632                elif incremental_sync.step or incremental_sync.cursor_granularity:
3633                    raise ValueError(
3634                        f"Found more that one slice per parent. LazySimpleRetriever only supports single slice read for stream - {name}."
3635                    )
3636
3637            if model.decoder and model.decoder.type != "JsonDecoder":
3638                raise ValueError(
3639                    f"LazySimpleRetriever only supports JsonDecoder. Found: {model.decoder.type}."
3640                )
3641
3642            return LazySimpleRetriever(
3643                name=name,
3644                paginator=paginator,
3645                primary_key=primary_key,
3646                requester=requester,
3647                record_selector=record_selector,
3648                stream_slicer=_NO_STREAM_SLICING,
3649                request_option_provider=request_options_provider,
3650                config=config,
3651                ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
3652                post_pagination_filter=post_pagination_filter,
3653                parameters=model.parameters or {},
3654            )
3655
3656        if (
3657            model.record_selector.record_filter
3658            and model.pagination_reset
3659            and model.pagination_reset.limits
3660        ):
3661            raise ValueError("PaginationResetLimits are not supported while having record filter.")
3662
3663        return SimpleRetriever(
3664            name=name,
3665            paginator=paginator,
3666            primary_key=primary_key,
3667            requester=requester,
3668            record_selector=record_selector,
3669            stream_slicer=_NO_STREAM_SLICING,
3670            request_option_provider=request_options_provider,
3671            config=config,
3672            ignore_stream_slicer_parameters_on_paginated_requests=ignore_stream_slicer_parameters_on_paginated_requests,
3673            additional_query_properties=query_properties,
3674            log_formatter=self._get_log_formatter(log_formatter, name),
3675            pagination_tracker_factory=self._create_pagination_tracker_factory(
3676                model.pagination_reset, cursor
3677            ),
3678            post_pagination_filter=post_pagination_filter,
3679            parameters=model.parameters or {},
3680        )
3681
3682    def _create_pagination_tracker_factory(
3683        self, model: Optional[PaginationResetModel], cursor: Cursor
3684    ) -> Callable[[], PaginationTracker]:
3685        if model is None:
3686            return lambda: PaginationTracker()
3687
3688        # Until we figure out a way to use any cursor for PaginationTracker, we will have to have this cursor selector logic
3689        cursor_factory: Callable[[], Optional[ConcurrentCursor]] = lambda: None
3690        if model.action == PaginationResetActionModel.RESET:
3691            # in that case, we will let cursor_factory to return None even if the stream has a cursor
3692            pass
3693        elif model.action == PaginationResetActionModel.SPLIT_USING_CURSOR:
3694            if isinstance(cursor, ConcurrentCursor):
3695                cursor_factory = lambda: cursor.copy_without_state()  # type: ignore  # the if condition validates that it is a ConcurrentCursor
3696            elif isinstance(cursor, ConcurrentPerPartitionCursor):
3697                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
3698                    {}, datetime.timedelta(0)
3699                )
3700            elif not isinstance(cursor, FinalStateCursor):
3701                LOGGER.warning(
3702                    "Unknown cursor for PaginationTracker. Pagination resets might not work properly"
3703                )
3704        else:
3705            raise ValueError(f"Unknown PaginationReset action: {model.action}")
3706
3707        limit = model.limits.number_of_records if model and model.limits else None
3708        return lambda: PaginationTracker(cursor_factory(), limit)
3709
3710    def _get_log_formatter(
3711        self, log_formatter: Callable[[Response], Any] | None, name: str
3712    ) -> Callable[[Response], Any] | None:
3713        if self._should_limit_slices_fetched():
3714            return (
3715                (
3716                    lambda response: format_http_message(
3717                        response,
3718                        f"Stream '{name}' request",
3719                        f"Request performed in order to extract records for stream '{name}'",
3720                        name,
3721                    )
3722                )
3723                if not log_formatter
3724                else log_formatter
3725            )
3726        return None
3727
3728    def _should_limit_slices_fetched(self) -> bool:
3729        """
3730        Returns True if the number of slices fetched should be limited, False otherwise.
3731        This is used to limit the number of slices fetched during tests.
3732        """
3733        return bool(self._limit_slices_fetched or self._emit_connector_builder_messages)
3734
3735    @staticmethod
3736    def _has_query_properties_in_request_parameters(
3737        requester: Union[HttpRequesterModel, CustomRequesterModel],
3738    ) -> bool:
3739        if not hasattr(requester, "request_parameters"):
3740            return False
3741        request_parameters = requester.request_parameters
3742        if request_parameters and isinstance(request_parameters, Mapping):
3743            for request_parameter in request_parameters.values():
3744                if isinstance(request_parameter, QueryPropertiesModel):
3745                    return True
3746        return False
3747
3748    @staticmethod
3749    def _remove_query_properties(
3750        request_parameters: Mapping[str, Union[str, QueryPropertiesModel]],
3751    ) -> Mapping[str, str]:
3752        return {
3753            parameter_field: request_parameter
3754            for parameter_field, request_parameter in request_parameters.items()
3755            if not isinstance(request_parameter, QueryPropertiesModel)
3756        }
3757
3758    def create_state_delegating_stream(
3759        self,
3760        model: StateDelegatingStreamModel,
3761        config: Config,
3762        **kwargs: Any,
3763    ) -> DefaultStream:
3764        if (
3765            model.full_refresh_stream.name != model.name
3766            or model.name != model.incremental_stream.name
3767        ):
3768            raise ValueError(
3769                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}."
3770            )
3771
3772        # Resolve api_retention_period with config context (supports Jinja2 interpolation)
3773        resolved_retention_period: Optional[str] = None
3774        if model.api_retention_period:
3775            interpolated_retention = InterpolatedString.create(
3776                model.api_retention_period, parameters=model.parameters or {}
3777            )
3778            resolved_value = interpolated_retention.eval(config=config)
3779            if resolved_value:
3780                resolved_retention_period = str(resolved_value)
3781
3782        if resolved_retention_period:
3783            for stream_model in (model.full_refresh_stream, model.incremental_stream):
3784                if isinstance(stream_model.incremental_sync, IncrementingCountCursorModel):
3785                    raise ValueError(
3786                        f"Stream '{model.name}' uses IncrementingCountCursor which is not supported "
3787                        f"with api_retention_period. IncrementingCountCursor does not use datetime-based "
3788                        f"cursors, so cursor age validation cannot be performed."
3789                    )
3790
3791        stream_state = self._connector_state_manager.get_stream_state(model.name, None)
3792
3793        if not stream_state:
3794            return self._create_component_from_model(  # type: ignore[no-any-return]
3795                model.full_refresh_stream, config=config, **kwargs
3796            )
3797
3798        incremental_stream: DefaultStream = self._create_component_from_model(
3799            model.incremental_stream, config=config, **kwargs
3800        )  # type: ignore[assignment]
3801
3802        # Only run cursor age validation for streams that are in the configured
3803        # catalog (or when no catalog was provided, e.g. during discover / connector
3804        # builder).  Streams not selected by the user but instantiated as parent-stream
3805        # dependencies must not go through this path because it emits state messages
3806        # that the destination does not know about, causing "Stream not found" crashes.
3807        stream_is_in_catalog = (
3808            not self._stream_name_to_configured_stream  # no catalog → validate by default
3809            or model.name in self._stream_name_to_configured_stream
3810        )
3811        if resolved_retention_period and stream_is_in_catalog:
3812            full_refresh_stream: DefaultStream = self._create_component_from_model(
3813                model.full_refresh_stream, config=config, **kwargs
3814            )  # type: ignore[assignment]
3815            if self._is_cursor_older_than_retention_period(
3816                stream_state,
3817                full_refresh_stream.cursor,
3818                incremental_stream.cursor,
3819                resolved_retention_period,
3820                model.name,
3821            ):
3822                # Clear state BEFORE constructing the full_refresh_stream so that
3823                # its cursor starts from start_date instead of the stale cursor.
3824                self._connector_state_manager.update_state_for_stream(model.name, None, {})
3825                state_message = self._connector_state_manager.create_state_message(model.name, None)
3826                self._message_repository.emit_message(state_message)
3827                return self._create_component_from_model(  # type: ignore[no-any-return]
3828                    model.full_refresh_stream, config=config, **kwargs
3829                )
3830
3831        return incremental_stream
3832
3833    @staticmethod
3834    def _is_cursor_older_than_retention_period(
3835        stream_state: Mapping[str, Any],
3836        full_refresh_cursor: Cursor,
3837        incremental_cursor: Cursor,
3838        api_retention_period: str,
3839        stream_name: str,
3840    ) -> bool:
3841        """Check if the cursor value in the state is older than the API's retention period.
3842
3843        Checks cursors in sequence: full refresh cursor first, then incremental cursor.
3844        FinalStateCursor returns now() for completed full refresh state (NO_CURSOR_STATE_KEY),
3845        which is always within retention, so we use incremental. For other states, it returns
3846        None and we fall back to checking the incremental cursor.
3847
3848        Returns True if the cursor is older than the retention period (should use full refresh).
3849        Returns False if the cursor is within the retention period (safe to use incremental).
3850        """
3851        retention_duration = parse_duration(api_retention_period)
3852        retention_cutoff = datetime.datetime.now(datetime.timezone.utc) - retention_duration
3853
3854        # Check full refresh cursor first
3855        cursor_datetime = full_refresh_cursor.get_cursor_datetime_from_state(stream_state)
3856
3857        # If full refresh cursor returns None, check incremental cursor
3858        if cursor_datetime is None:
3859            cursor_datetime = incremental_cursor.get_cursor_datetime_from_state(stream_state)
3860
3861        if cursor_datetime is None:
3862            # Neither cursor could parse the state - fall back to full refresh to be safe
3863            return True
3864
3865        if cursor_datetime < retention_cutoff:
3866            logging.warning(
3867                f"Stream '{stream_name}' has a cursor value older than "
3868                f"the API's retention period of {api_retention_period} "
3869                f"(cutoff: {retention_cutoff.isoformat()}). "
3870                f"Falling back to full refresh to avoid data loss."
3871            )
3872            return True
3873
3874        return False
3875
3876    def _get_state_delegating_stream_model(
3877        self,
3878        model: StateDelegatingStreamModel,
3879        parent_state: Optional[Mapping[str, Any]] = None,
3880    ) -> DeclarativeStreamModel:
3881        """Return the appropriate underlying stream model based on state."""
3882        return (
3883            model.incremental_stream
3884            if self._connector_state_manager.get_stream_state(model.name, None) or parent_state
3885            else model.full_refresh_stream
3886        )
3887
3888    _OPTIONAL_ASYNC_STATUS_FIELDS = {"skipped"}
3889
3890    def _create_async_job_status_mapping(
3891        self, model: AsyncJobStatusMapModel, config: Config, **kwargs: Any
3892    ) -> Mapping[str, AsyncJobStatus]:
3893        api_status_to_cdk_status = {}
3894        for cdk_status, api_statuses in model.dict().items():
3895            if cdk_status == "type":
3896                # This is an element of the dict because of the typing of the CDK but it is not a CDK status
3897                continue
3898
3899            if api_statuses is None:
3900                if cdk_status in self._OPTIONAL_ASYNC_STATUS_FIELDS:
3901                    continue
3902                raise ValueError(
3903                    f"Required CDK status '{cdk_status}' has no API statuses mapped. "
3904                    f"Please provide at least an empty list for required status fields."
3905                )
3906
3907            for status in api_statuses:
3908                if status in api_status_to_cdk_status:
3909                    raise ValueError(
3910                        f"API status {status} is already set for CDK status {cdk_status}. Please ensure API statuses are only provided once"
3911                    )
3912                api_status_to_cdk_status[status] = self._get_async_job_status(cdk_status)
3913        return api_status_to_cdk_status
3914
3915    def _get_async_job_status(self, status: str) -> AsyncJobStatus:
3916        match status:
3917            case "running":
3918                return AsyncJobStatus.RUNNING
3919            case "completed":
3920                return AsyncJobStatus.COMPLETED
3921            case "failed":
3922                return AsyncJobStatus.FAILED
3923            case "timeout":
3924                return AsyncJobStatus.TIMED_OUT
3925            case "skipped":
3926                return AsyncJobStatus.SKIPPED
3927            case _:
3928                raise ValueError(f"Unsupported CDK status {status}")
3929
3930    def create_async_retriever(
3931        self,
3932        model: AsyncRetrieverModel,
3933        config: Config,
3934        *,
3935        name: str,
3936        primary_key: Optional[
3937            Union[str, List[str], List[List[str]]]
3938        ],  # this seems to be needed to match create_simple_retriever
3939        stream_slicer: Optional[StreamSlicer],
3940        client_side_incremental_sync: Optional[Dict[str, Any]] = None,
3941        transformations: List[RecordTransformation],
3942        **kwargs: Any,
3943    ) -> AsyncRetriever:
3944        if model.download_target_requester and not model.download_target_extractor:
3945            raise ValueError(
3946                f"`download_target_extractor` required if using a `download_target_requester`"
3947            )
3948
3949        def _get_download_retriever(
3950            requester: Requester, extractor: RecordExtractor, _decoder: Decoder
3951        ) -> SimpleRetriever:
3952            # We create a record selector for the download retriever
3953            # with no schema normalization and no transformations, neither record filter
3954            # as all this occurs in the record_selector of the AsyncRetriever
3955            record_selector = RecordSelector(
3956                extractor=extractor,
3957                name=name,
3958                record_filter=None,
3959                transformations=[],
3960                schema_normalization=TypeTransformer(TransformConfig.NoTransform),
3961                config=config,
3962                parameters={},
3963            )
3964            paginator = (
3965                self._create_component_from_model(
3966                    model=model.download_paginator,
3967                    decoder=_decoder,
3968                    config=config,
3969                    url_base="",
3970                )
3971                if model.download_paginator
3972                else NoPagination(parameters={})
3973            )
3974
3975            return SimpleRetriever(
3976                requester=requester,
3977                record_selector=record_selector,
3978                primary_key=None,
3979                name=name,
3980                paginator=paginator,
3981                config=config,
3982                parameters={},
3983                log_formatter=self._get_log_formatter(None, name),
3984            )
3985
3986        def _get_job_timeout() -> datetime.timedelta:
3987            user_defined_timeout: Optional[int] = (
3988                int(
3989                    InterpolatedString.create(
3990                        str(model.polling_job_timeout),
3991                        parameters={},
3992                    ).eval(config)
3993                )
3994                if model.polling_job_timeout
3995                else None
3996            )
3997
3998            # check for user defined timeout during the test read or 15 minutes
3999            test_read_timeout = datetime.timedelta(minutes=user_defined_timeout or 15)
4000            # default value for non-connector builder is 60 minutes.
4001            default_sync_timeout = datetime.timedelta(minutes=user_defined_timeout or 60)
4002
4003            return (
4004                test_read_timeout if self._emit_connector_builder_messages else default_sync_timeout
4005            )
4006
4007        decoder = (
4008            self._create_component_from_model(model=model.decoder, config=config)
4009            if model.decoder
4010            else JsonDecoder(parameters={})
4011        )
4012        record_selector = self._create_component_from_model(
4013            model=model.record_selector,
4014            config=config,
4015            decoder=decoder,
4016            name=name,
4017            transformations=transformations,
4018            client_side_incremental_sync=client_side_incremental_sync,
4019        )
4020
4021        stream_slicer = stream_slicer or SinglePartitionRouter(parameters={})
4022        if self._should_limit_slices_fetched():
4023            stream_slicer = cast(
4024                StreamSlicer,
4025                StreamSlicerTestReadDecorator(
4026                    wrapped_slicer=stream_slicer,
4027                    maximum_number_of_slices=self._limit_slices_fetched or 5,
4028                ),
4029            )
4030
4031        creation_requester = self._create_component_from_model(
4032            model=model.creation_requester,
4033            decoder=decoder,
4034            config=config,
4035            name=f"job creation - {name}",
4036        )
4037        polling_requester = self._create_component_from_model(
4038            model=model.polling_requester,
4039            decoder=decoder,
4040            config=config,
4041            name=f"job polling - {name}",
4042        )
4043        job_download_components_name = f"job download - {name}"
4044        download_decoder = (
4045            self._create_component_from_model(model=model.download_decoder, config=config)
4046            if model.download_decoder
4047            else JsonDecoder(parameters={})
4048        )
4049        download_extractor = (
4050            self._create_component_from_model(
4051                model=model.download_extractor,
4052                config=config,
4053                decoder=download_decoder,
4054                parameters=model.parameters,
4055            )
4056            if model.download_extractor
4057            else DpathExtractor(
4058                [],
4059                config=config,
4060                decoder=download_decoder,
4061                parameters=model.parameters or {},
4062            )
4063        )
4064        download_requester = self._create_component_from_model(
4065            model=model.download_requester,
4066            decoder=download_decoder,
4067            config=config,
4068            name=job_download_components_name,
4069        )
4070        download_retriever = _get_download_retriever(
4071            download_requester, download_extractor, download_decoder
4072        )
4073        abort_requester = (
4074            self._create_component_from_model(
4075                model=model.abort_requester,
4076                decoder=decoder,
4077                config=config,
4078                name=f"job abort - {name}",
4079            )
4080            if model.abort_requester
4081            else None
4082        )
4083        delete_requester = (
4084            self._create_component_from_model(
4085                model=model.delete_requester,
4086                decoder=decoder,
4087                config=config,
4088                name=f"job delete - {name}",
4089            )
4090            if model.delete_requester
4091            else None
4092        )
4093        download_target_requester = (
4094            self._create_component_from_model(
4095                model=model.download_target_requester,
4096                decoder=decoder,
4097                config=config,
4098                name=f"job extract_url - {name}",
4099            )
4100            if model.download_target_requester
4101            else None
4102        )
4103        status_extractor = self._create_component_from_model(
4104            model=model.status_extractor, decoder=decoder, config=config, name=name
4105        )
4106        download_target_extractor = (
4107            self._create_component_from_model(
4108                model=model.download_target_extractor,
4109                decoder=decoder,
4110                config=config,
4111                name=name,
4112            )
4113            if model.download_target_extractor
4114            else None
4115        )
4116
4117        job_repository: AsyncJobRepository = AsyncHttpJobRepository(
4118            creation_requester=creation_requester,
4119            polling_requester=polling_requester,
4120            download_retriever=download_retriever,
4121            download_target_requester=download_target_requester,
4122            abort_requester=abort_requester,
4123            delete_requester=delete_requester,
4124            status_extractor=status_extractor,
4125            status_mapping=self._create_async_job_status_mapping(model.status_mapping, config),
4126            download_target_extractor=download_target_extractor,
4127            job_timeout=_get_job_timeout(),
4128        )
4129
4130        failed_retry_wait_time_in_seconds: Optional[int] = (
4131            int(
4132                InterpolatedString.create(
4133                    str(model.failed_retry_wait_time_in_seconds),
4134                    parameters={},
4135                ).eval(config)
4136            )
4137            if model.failed_retry_wait_time_in_seconds
4138            else None
4139        )
4140
4141        async_job_partition_router = AsyncJobPartitionRouter(
4142            job_orchestrator_factory=lambda stream_slices: AsyncJobOrchestrator(
4143                job_repository,
4144                stream_slices,
4145                self._job_tracker,
4146                self._message_repository,
4147                # FIXME work would need to be done here in order to detect if a stream as a parent stream that is bulk
4148                has_bulk_parent=False,
4149                # set the `job_max_retry` to 1 for the `Connector Builder`` use-case.
4150                # `None` == default retry is set to 3 attempts, under the hood.
4151                job_max_retry=1 if self._emit_connector_builder_messages else None,
4152                failed_retry_wait_time_in_seconds=failed_retry_wait_time_in_seconds,
4153            ),
4154            stream_slicer=stream_slicer,
4155            config=config,
4156            parameters=model.parameters or {},
4157        )
4158
4159        return AsyncRetriever(
4160            record_selector=record_selector,
4161            stream_slicer=async_job_partition_router,
4162            config=config,
4163            parameters=model.parameters or {},
4164        )
4165
4166    def create_spec(self, model: SpecModel, config: Config, **kwargs: Any) -> Spec:
4167        config_migrations = [
4168            self._create_component_from_model(migration, config)
4169            for migration in (
4170                model.config_normalization_rules.config_migrations
4171                if (
4172                    model.config_normalization_rules
4173                    and model.config_normalization_rules.config_migrations
4174                )
4175                else []
4176            )
4177        ]
4178        config_transformations = [
4179            self._create_component_from_model(transformation, config)
4180            for transformation in (
4181                model.config_normalization_rules.transformations
4182                if (
4183                    model.config_normalization_rules
4184                    and model.config_normalization_rules.transformations
4185                )
4186                else []
4187            )
4188        ]
4189        config_validations = [
4190            self._create_component_from_model(validation, config)
4191            for validation in (
4192                model.config_normalization_rules.validations
4193                if (
4194                    model.config_normalization_rules
4195                    and model.config_normalization_rules.validations
4196                )
4197                else []
4198            )
4199        ]
4200
4201        return Spec(
4202            connection_specification=model.connection_specification,
4203            documentation_url=model.documentation_url,
4204            advanced_auth=model.advanced_auth,
4205            parameters={},
4206            config_migrations=config_migrations,
4207            config_transformations=config_transformations,
4208            config_validations=config_validations,
4209        )
4210
4211    def create_substream_partition_router(
4212        self,
4213        model: SubstreamPartitionRouterModel,
4214        config: Config,
4215        *,
4216        stream_name: str,
4217        **kwargs: Any,
4218    ) -> SubstreamPartitionRouter:
4219        parent_stream_configs = []
4220        if model.parent_stream_configs:
4221            parent_stream_configs.extend(
4222                [
4223                    self.create_parent_stream_config_with_substream_wrapper(
4224                        model=parent_stream_config, config=config, stream_name=stream_name, **kwargs
4225                    )
4226                    for parent_stream_config in model.parent_stream_configs
4227                ]
4228            )
4229
4230        return SubstreamPartitionRouter(
4231            parent_stream_configs=parent_stream_configs,
4232            parameters=model.parameters or {},
4233            config=config,
4234        )
4235
4236    def create_parent_stream_config_with_substream_wrapper(
4237        self, model: ParentStreamConfigModel, config: Config, *, stream_name: str, **kwargs: Any
4238    ) -> Any:
4239        child_state = self._connector_state_manager.get_stream_state(stream_name, None)
4240        if NO_CURSOR_STATE_KEY in child_state:
4241            # Full refresh streams checkpoint a `{NO_CURSOR_STATE_KEY: true}` sentinel. When such a
4242            # stream is later converted to incremental with an incremental_dependency parent,
4243            # `_instantiate_parent_stream_state_manager` would treat the sentinel's boolean as a legacy
4244            # cursor value and re-key it under the parent's cursor field, crashing cursor initialization.
4245            child_state = {
4246                key: value for key, value in child_state.items() if key != NO_CURSOR_STATE_KEY
4247            }
4248
4249        parent_state: Optional[Mapping[str, Any]] = (
4250            child_state if model.incremental_dependency and child_state else None
4251        )
4252        connector_state_manager = self._instantiate_parent_stream_state_manager(
4253            child_state, config, model, parent_state
4254        )
4255
4256        substream_factory = ModelToComponentFactory(
4257            custom_components_trusted=self._custom_components_trusted,
4258            connector_state_manager=connector_state_manager,
4259            limit_pages_fetched_per_slice=self._limit_pages_fetched_per_slice,
4260            limit_slices_fetched=self._limit_slices_fetched,
4261            emit_connector_builder_messages=self._emit_connector_builder_messages,
4262            disable_retries=self._disable_retries,
4263            disable_cache=self._disable_cache,
4264            message_repository=StateFilteringMessageRepository(
4265                LogAppenderMessageRepositoryDecorator(
4266                    {
4267                        "airbyte_cdk": {"stream": {"is_substream": True}},
4268                        "http": {"is_auxiliary": True},
4269                    },
4270                    self._message_repository,
4271                    self._evaluate_log_level(self._emit_connector_builder_messages),
4272                ),
4273            ),
4274            api_budget=self._api_budget,
4275            # Share the authenticator registry so parent and child streams draw from the
4276            # same token quota counters
4277            rate_limited_authenticators=self._rate_limited_authenticators,
4278        )
4279
4280        return substream_factory.create_parent_stream_config(
4281            model=model, config=config, stream_name=stream_name, **kwargs
4282        )
4283
4284    def _instantiate_parent_stream_state_manager(
4285        self,
4286        child_state: MutableMapping[str, Any],
4287        config: Config,
4288        model: ParentStreamConfigModel,
4289        parent_state: Optional[Mapping[str, Any]] = None,
4290    ) -> ConnectorStateManager:
4291        """
4292        With DefaultStream, the state needs to be provided during __init__ of the cursor as opposed to the
4293        `set_initial_state` flow that existed for the declarative cursors. This state is taken from
4294        self._connector_state_manager.get_stream_state (`self` being a newly created ModelToComponentFactory to account
4295        for the MessageRepository being different). So we need to pass a ConnectorStateManager to the
4296        ModelToComponentFactory that has the parent states. This method populates this if there is a child state and if
4297        incremental_dependency is set.
4298        """
4299        if model.incremental_dependency and child_state:
4300            parent_stream_name = model.stream.name or ""
4301            extracted_parent_state = ConcurrentPerPartitionCursor.get_parent_state(
4302                child_state, parent_stream_name
4303            )
4304
4305            if not extracted_parent_state:
4306                extracted_parent_state = ConcurrentPerPartitionCursor.get_global_state(
4307                    child_state, parent_stream_name
4308                )
4309
4310                if not extracted_parent_state and not isinstance(extracted_parent_state, dict):
4311                    cursor_values = child_state.values()
4312                    if cursor_values and len(cursor_values) == 1:
4313                        incremental_sync_model: Union[
4314                            DatetimeBasedCursorModel,
4315                            IncrementingCountCursorModel,
4316                        ] = (
4317                            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
4318                            if isinstance(model.stream, DeclarativeStreamModel)
4319                            else self._get_state_delegating_stream_model(
4320                                model.stream, parent_state=parent_state
4321                            ).incremental_sync
4322                        )
4323                        cursor_field = InterpolatedString.create(
4324                            incremental_sync_model.cursor_field,
4325                            parameters=incremental_sync_model.parameters or {},
4326                        ).eval(config)
4327                        extracted_parent_state = AirbyteStateMessage(
4328                            type=AirbyteStateType.STREAM,
4329                            stream=AirbyteStreamState(
4330                                stream_descriptor=StreamDescriptor(
4331                                    name=parent_stream_name, namespace=None
4332                                ),
4333                                stream_state=AirbyteStateBlob(
4334                                    {cursor_field: list(cursor_values)[0]}
4335                                ),
4336                            ),
4337                        )
4338            return ConnectorStateManager([extracted_parent_state] if extracted_parent_state else [])
4339
4340        return ConnectorStateManager([])
4341
4342    @staticmethod
4343    def create_wait_time_from_header(
4344        model: WaitTimeFromHeaderModel, config: Config, **kwargs: Any
4345    ) -> WaitTimeFromHeaderBackoffStrategy:
4346        return WaitTimeFromHeaderBackoffStrategy(
4347            header=model.header,
4348            parameters=model.parameters or {},
4349            config=config,
4350            regex=model.regex,
4351            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
4352        )
4353
4354    @staticmethod
4355    def create_wait_until_time_from_header(
4356        model: WaitUntilTimeFromHeaderModel, config: Config, **kwargs: Any
4357    ) -> WaitUntilTimeFromHeaderBackoffStrategy:
4358        return WaitUntilTimeFromHeaderBackoffStrategy(
4359            header=model.header,
4360            parameters=model.parameters or {},
4361            config=config,
4362            min_wait=model.min_wait,
4363            regex=model.regex,
4364            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
4365        )
4366
4367    def get_message_repository(self) -> MessageRepository:
4368        return self._message_repository
4369
4370    def _evaluate_log_level(self, emit_connector_builder_messages: bool) -> Level:
4371        return Level.DEBUG if emit_connector_builder_messages else Level.INFO
4372
4373    @staticmethod
4374    def create_components_mapping_definition(
4375        model: ComponentMappingDefinitionModel, config: Config, **kwargs: Any
4376    ) -> ComponentMappingDefinition:
4377        interpolated_value = InterpolatedString.create(
4378            model.value, parameters=model.parameters or {}
4379        )
4380        field_path = [
4381            InterpolatedString.create(path, parameters=model.parameters or {})
4382            for path in model.field_path
4383        ]
4384        return ComponentMappingDefinition(
4385            field_path=field_path,  # type: ignore[arg-type] # field_path can be str and InterpolatedString
4386            value=interpolated_value,
4387            value_type=ModelToComponentFactory._json_schema_type_name_to_type(model.value_type),
4388            create_or_update=model.create_or_update,
4389            condition=model.condition,
4390            parameters=model.parameters or {},
4391        )
4392
4393    def create_http_components_resolver(
4394        self, model: HttpComponentsResolverModel, config: Config, stream_name: Optional[str] = None
4395    ) -> Any:
4396        retriever = self._create_component_from_model(
4397            model=model.retriever,
4398            config=config,
4399            name=f"{stream_name if stream_name else '__http_components_resolver'}",
4400            primary_key=None,
4401            stream_slicer=self._build_stream_slicer_from_partition_router(model.retriever, config),
4402            transformations=[],
4403        )
4404
4405        components_mapping = []
4406        for component_mapping_definition_model in model.components_mapping:
4407            if component_mapping_definition_model.condition:
4408                raise ValueError("`condition` is only supported for     `ConfigComponentsResolver`")
4409            components_mapping.append(
4410                self._create_component_from_model(
4411                    model=component_mapping_definition_model,
4412                    value_type=ModelToComponentFactory._json_schema_type_name_to_type(
4413                        component_mapping_definition_model.value_type
4414                    ),
4415                    config=config,
4416                )
4417            )
4418
4419        return HttpComponentsResolver(
4420            retriever=retriever,
4421            stream_slicer=self._build_stream_slicer_from_partition_router(model.retriever, config),
4422            config=config,
4423            components_mapping=components_mapping,
4424            parameters=model.parameters or {},
4425        )
4426
4427    @staticmethod
4428    def create_stream_config(
4429        model: StreamConfigModel, config: Config, **kwargs: Any
4430    ) -> StreamConfig:
4431        model_configs_pointer: List[Union[InterpolatedString, str]] = (
4432            [x for x in model.configs_pointer] if model.configs_pointer else []
4433        )
4434
4435        return StreamConfig(
4436            configs_pointer=model_configs_pointer,
4437            default_values=model.default_values,
4438            parameters=model.parameters or {},
4439        )
4440
4441    def create_config_components_resolver(
4442        self,
4443        model: ConfigComponentsResolverModel,
4444        config: Config,
4445    ) -> Any:
4446        model_stream_configs = (
4447            model.stream_config if isinstance(model.stream_config, list) else [model.stream_config]
4448        )
4449
4450        stream_configs = [
4451            self._create_component_from_model(
4452                stream_config, config=config, parameters=model.parameters or {}
4453            )
4454            for stream_config in model_stream_configs
4455        ]
4456
4457        components_mapping = [
4458            self._create_component_from_model(
4459                model=components_mapping_definition_model,
4460                value_type=ModelToComponentFactory._json_schema_type_name_to_type(
4461                    components_mapping_definition_model.value_type
4462                ),
4463                config=config,
4464                parameters=model.parameters,
4465            )
4466            for components_mapping_definition_model in model.components_mapping
4467        ]
4468
4469        return ConfigComponentsResolver(
4470            stream_configs=stream_configs,
4471            config=config,
4472            components_mapping=components_mapping,
4473            parameters=model.parameters or {},
4474        )
4475
4476    def create_parametrized_components_resolver(
4477        self,
4478        model: ParametrizedComponentsResolverModel,
4479        config: Config,
4480    ) -> ParametrizedComponentsResolver:
4481        stream_parameters = StreamParametersDefinition(
4482            list_of_parameters_for_stream=model.stream_parameters.list_of_parameters_for_stream
4483        )
4484
4485        components_mapping = []
4486        for components_mapping_definition_model in model.components_mapping:
4487            if components_mapping_definition_model.condition:
4488                raise ValueError("`condition` is only supported for `ConfigComponentsResolver`")
4489            components_mapping.append(
4490                self._create_component_from_model(
4491                    model=components_mapping_definition_model,
4492                    value_type=ModelToComponentFactory._json_schema_type_name_to_type(
4493                        components_mapping_definition_model.value_type
4494                    ),
4495                    config=config,
4496                )
4497            )
4498        return ParametrizedComponentsResolver(
4499            stream_parameters=stream_parameters,
4500            config=config,
4501            components_mapping=components_mapping,
4502            parameters=model.parameters or {},
4503        )
4504
4505    _UNSUPPORTED_DECODER_ERROR = (
4506        "Specified decoder of {decoder_type} is not supported for pagination."
4507        "Please set as `JsonDecoder`, `XmlDecoder`, or a `CompositeRawDecoder` with an inner_parser of `JsonParser` or `GzipParser` instead."
4508        "If using `GzipParser`, please ensure that the lowest level inner_parser is a `JsonParser`."
4509    )
4510
4511    def _is_supported_decoder_for_pagination(self, decoder: Decoder) -> bool:
4512        if isinstance(decoder, (JsonDecoder, XmlDecoder)):
4513            return True
4514        elif isinstance(decoder, CompositeRawDecoder):
4515            return self._is_supported_parser_for_pagination(decoder.parser)
4516        else:
4517            return False
4518
4519    def _is_supported_parser_for_pagination(self, parser: Parser) -> bool:
4520        if isinstance(parser, JsonParser):
4521            return True
4522        elif isinstance(parser, GzipParser):
4523            return isinstance(parser.inner_parser, JsonParser)
4524        else:
4525            return False
4526
4527    def create_http_api_budget(
4528        self, model: HTTPAPIBudgetModel, config: Config, **kwargs: Any
4529    ) -> HttpAPIBudget:
4530        policies = [
4531            self._create_component_from_model(model=policy, config=config)
4532            for policy in model.policies
4533        ]
4534
4535        return HttpAPIBudget(
4536            policies=policies,
4537            ratelimit_reset_header=model.ratelimit_reset_header or "ratelimit-reset",
4538            ratelimit_remaining_header=model.ratelimit_remaining_header or "ratelimit-remaining",
4539            status_codes_for_ratelimit_hit=model.status_codes_for_ratelimit_hit or [429],
4540        )
4541
4542    def create_fixed_window_call_rate_policy(
4543        self, model: FixedWindowCallRatePolicyModel, config: Config, **kwargs: Any
4544    ) -> FixedWindowCallRatePolicy:
4545        matchers = [
4546            self._create_component_from_model(model=matcher, config=config)
4547            for matcher in model.matchers
4548        ]
4549
4550        # Set the initial reset timestamp to 10 days from now.
4551        # This value will be updated by the first request.
4552        return FixedWindowCallRatePolicy(
4553            next_reset_ts=datetime.datetime.now() + datetime.timedelta(days=10),
4554            period=parse_duration(model.period),
4555            call_limit=model.call_limit,
4556            matchers=matchers,
4557        )
4558
4559    def create_file_uploader(
4560        self, model: FileUploaderModel, config: Config, **kwargs: Any
4561    ) -> FileUploader:
4562        name = "File Uploader"
4563        requester = self._create_component_from_model(
4564            model=model.requester,
4565            config=config,
4566            name=name,
4567            **kwargs,
4568        )
4569        download_target_extractor = self._create_component_from_model(
4570            model=model.download_target_extractor,
4571            config=config,
4572            name=name,
4573            **kwargs,
4574        )
4575        emit_connector_builder_messages = self._emit_connector_builder_messages
4576        file_uploader = DefaultFileUploader(
4577            requester=requester,
4578            download_target_extractor=download_target_extractor,
4579            config=config,
4580            file_writer=NoopFileWriter()
4581            if emit_connector_builder_messages
4582            else LocalFileSystemFileWriter(),
4583            parameters=model.parameters or {},
4584            filename_extractor=model.filename_extractor if model.filename_extractor else None,
4585        )
4586
4587        return (
4588            ConnectorBuilderFileUploader(file_uploader)
4589            if emit_connector_builder_messages
4590            else file_uploader
4591        )
4592
4593    def create_moving_window_call_rate_policy(
4594        self, model: MovingWindowCallRatePolicyModel, config: Config, **kwargs: Any
4595    ) -> MovingWindowCallRatePolicy:
4596        rates = [
4597            self._create_component_from_model(model=rate, config=config) for rate in model.rates
4598        ]
4599        matchers = [
4600            self._create_component_from_model(model=matcher, config=config)
4601            for matcher in model.matchers
4602        ]
4603        return MovingWindowCallRatePolicy(
4604            rates=rates,
4605            matchers=matchers,
4606        )
4607
4608    def create_unlimited_call_rate_policy(
4609        self, model: UnlimitedCallRatePolicyModel, config: Config, **kwargs: Any
4610    ) -> UnlimitedCallRatePolicy:
4611        matchers = [
4612            self._create_component_from_model(model=matcher, config=config)
4613            for matcher in model.matchers
4614        ]
4615
4616        return UnlimitedCallRatePolicy(
4617            matchers=matchers,
4618        )
4619
4620    def create_rate(self, model: RateModel, config: Config, **kwargs: Any) -> Rate:
4621        interpolated_limit = InterpolatedString.create(str(model.limit), parameters={})
4622        return Rate(
4623            limit=int(interpolated_limit.eval(config=config)),
4624            interval=parse_duration(model.interval),
4625        )
4626
4627    def create_http_request_matcher(
4628        self, model: HttpRequestRegexMatcherModel, config: Config, **kwargs: Any
4629    ) -> HttpRequestRegexMatcher:
4630        weight = model.weight
4631        if weight is not None:
4632            if isinstance(weight, str):
4633                weight = int(InterpolatedString.create(weight, parameters={}).eval(config))
4634            else:
4635                weight = int(weight)
4636            if weight < 1:
4637                raise ValueError(f"weight must be >= 1, got {weight}")
4638        return HttpRequestRegexMatcher(
4639            method=model.method,
4640            url_base=model.url_base,
4641            url_path_pattern=model.url_path_pattern,
4642            params=model.params,
4643            headers=model.headers,
4644            weight=weight,
4645        )
4646
4647    def create_rate_limited_multiple_token_authenticator(
4648        self,
4649        model: RateLimitedMultipleTokenAuthenticatorModel,
4650        config: Config,
4651        **kwargs: Any,
4652    ) -> RateLimitedMultipleTokenAuthenticator:
4653        if isinstance(model.tokens, str):
4654            tokens_value = InterpolatedString.create(model.tokens, parameters={}).eval(config)
4655            delimiter = model.token_delimiter or ","
4656            tokens = [
4657                token.strip() for token in str(tokens_value).split(delimiter) if token.strip()
4658            ]
4659        else:
4660            tokens = [
4661                token_value
4662                for token in model.tokens
4663                if (
4664                    token_value := str(
4665                        InterpolatedString.create(token, parameters={}).eval(config)
4666                    ).strip()
4667                )
4668            ]
4669
4670        quota_specs = [
4671            {
4672                "name": quota_model.name,
4673                "remaining_path": quota_model.remaining_path,
4674                "reset_path": quota_model.reset_path,
4675                "limit_path": quota_model.limit_path,
4676                "remaining_header": quota_model.remaining_header,
4677                "reset_header": quota_model.reset_header,
4678                "limit_header": quota_model.limit_header,
4679                # Normalize the same way as the runtime TokenQuota below, so an omitted field
4680                # and an explicit `[]` key identically and keep sharing one set of counters.
4681                "exhaustion_status_codes": quota_model.exhaustion_status_codes or [],
4682                "matchers": [
4683                    {
4684                        "method": matcher_model.method,
4685                        "url_base": matcher_model.url_base,
4686                        "url_path_pattern": matcher_model.url_path_pattern,
4687                        "params": matcher_model.params,
4688                        "headers": matcher_model.headers,
4689                        "weight": matcher_model.weight,
4690                    }
4691                    for matcher_model in quota_model.matchers or []
4692                ],
4693            }
4694            for quota_model in model.quotas
4695        ]
4696
4697        quota_status_url = str(
4698            InterpolatedString.create(model.quota_status_source.url, parameters={}).eval(config)
4699        )
4700        quota_status_http_method = (
4701            model.quota_status_source.http_method.value
4702            if model.quota_status_source.http_method
4703            else "GET"
4704        )
4705        quota_status_headers = {
4706            key: str(InterpolatedString.create(value, parameters={}).eval(config))
4707            for key, value in (model.quota_status_source.request_headers or {}).items()
4708        }
4709        # Normalize the same way as the quota specs above, so an omitted field and an explicit
4710        # `[]` key identically and keep sharing one set of counters. Deduplicated as well as
4711        # sorted, because the runtime turns this into a set: without it `[404]` and `[404, 404]`
4712        # would key differently and stop sharing counters while behaving identically.
4713        quota_status_unavailable_status_codes = sorted(
4714            set(model.quota_status_source.unavailable_status_codes or [])
4715        )
4716        auth_method = model.auth_method or "Bearer"
4717        header = model.header or "Authorization"
4718        max_wait_time_str = str(
4719            InterpolatedString.create(model.max_wait_time or "PT2H", parameters={}).eval(config)
4720        )
4721        max_wait_time = parse_duration(max_wait_time_str)
4722        if not isinstance(max_wait_time, datetime.timedelta):
4723            raise ValueError(
4724                f"max_wait_time must be a fixed-length ISO 8601 duration (e.g. 'PT2H'); "
4725                f"calendar-unit durations like '{max_wait_time_str}' are not supported"
4726            )
4727        budget_reserve_fraction = (
4728            model.budget_reserve_fraction if model.budget_reserve_fraction is not None else 0.1
4729        )
4730        budget_min_reserve = (
4731            model.budget_min_reserve if model.budget_min_reserve is not None else 50
4732        )
4733
4734        # Reuse the same instance for identical definitions so that all streams share the
4735        # same token quota counters (similar to how api_budget is shared). The key is built
4736        # from the resolved constructor arguments rather than the raw model so that
4737        # stream-specific `$parameters` propagated onto the model (and its nested components)
4738        # cannot break instance sharing.
4739        cache_key = json.dumps(
4740            {
4741                "tokens": tokens,
4742                "quotas": quota_specs,
4743                "quota_status_url": quota_status_url,
4744                "quota_status_http_method": quota_status_http_method,
4745                "quota_status_headers": quota_status_headers,
4746                "quota_status_unavailable_status_codes": quota_status_unavailable_status_codes,
4747                "auth_method": auth_method,
4748                "header": header,
4749                "max_wait_time": max_wait_time.total_seconds(),
4750                "budget_reserve_fraction": budget_reserve_fraction,
4751                "budget_min_reserve": budget_min_reserve,
4752            },
4753            sort_keys=True,
4754        )
4755        if cache_key in self._rate_limited_authenticators:
4756            return self._rate_limited_authenticators[cache_key]
4757
4758        quotas = [
4759            TokenQuota(
4760                name=quota_model.name,
4761                remaining_path=quota_model.remaining_path,
4762                reset_path=quota_model.reset_path,
4763                limit_path=quota_model.limit_path,
4764                remaining_header=quota_model.remaining_header,
4765                reset_header=quota_model.reset_header,
4766                limit_header=quota_model.limit_header,
4767                exhaustion_status_codes=quota_model.exhaustion_status_codes or [],
4768                matchers=[
4769                    self.create_http_request_matcher(matcher_model, config)
4770                    for matcher_model in quota_model.matchers or []
4771                ],
4772            )
4773            for quota_model in model.quotas
4774        ]
4775
4776        authenticator = RateLimitedMultipleTokenAuthenticator(
4777            tokens=tokens,
4778            quotas=quotas,
4779            quota_status_url=quota_status_url,
4780            quota_status_http_method=quota_status_http_method,
4781            quota_status_headers=quota_status_headers,
4782            quota_status_unavailable_status_codes=quota_status_unavailable_status_codes,
4783            auth_method=auth_method,
4784            header=header,
4785            max_wait_time=max_wait_time,
4786            budget_reserve_fraction=budget_reserve_fraction,
4787            budget_min_reserve=budget_min_reserve,
4788        )
4789        self._rate_limited_authenticators[cache_key] = authenticator
4790        return authenticator
4791
4792    def set_api_budget(self, component_definition: ComponentDefinition, config: Config) -> None:
4793        self._api_budget = self.create_component(
4794            model_type=HTTPAPIBudgetModel, component_definition=component_definition, config=config
4795        )
4796
4797    def create_grouping_partition_router(
4798        self,
4799        model: GroupingPartitionRouterModel,
4800        config: Config,
4801        *,
4802        stream_name: str,
4803        **kwargs: Any,
4804    ) -> GroupingPartitionRouter:
4805        underlying_router = self._create_component_from_model(
4806            model=model.underlying_partition_router,
4807            config=config,
4808            stream_name=stream_name,
4809            **kwargs,
4810        )
4811        if model.group_size < 1:
4812            raise ValueError(f"Group size must be greater than 0, got {model.group_size}")
4813
4814        # Request options in underlying partition routers are not supported for GroupingPartitionRouter
4815        # because they are specific to individual partitions and cannot be aggregated or handled
4816        # when grouping, potentially leading to incorrect API calls. Any request customization
4817        # should be managed at the stream level through the requester's configuration.
4818        if isinstance(underlying_router, SubstreamPartitionRouter):
4819            if any(
4820                parent_config.request_option
4821                for parent_config in underlying_router.parent_stream_configs
4822            ):
4823                raise ValueError("Request options are not supported for GroupingPartitionRouter.")
4824
4825        if isinstance(underlying_router, ListPartitionRouter):
4826            if underlying_router.request_option:
4827                raise ValueError("Request options are not supported for GroupingPartitionRouter.")
4828
4829        return GroupingPartitionRouter(
4830            group_size=model.group_size,
4831            underlying_partition_router=underlying_router,
4832            deduplicate=model.deduplicate if model.deduplicate is not None else True,
4833            config=config,
4834        )
4835
4836    def create_union_partition_router(
4837        self,
4838        model: UnionPartitionRouterModel,
4839        config: Config,
4840        *,
4841        stream_name: str,
4842        **kwargs: Any,
4843    ) -> UnionPartitionRouter:
4844        # The schema enforces minItems: 2 for manifests; this guard covers construction paths
4845        # that bypass JSON-schema validation (the generated model carries no min_items constraint).
4846        if len(model.partition_routers) < 2:
4847            raise ValueError(
4848                f"UnionPartitionRouter for stream {stream_name} needs at least 2 child partition routers"
4849            )
4850
4851        partition_routers = [
4852            self._create_component_from_model(
4853                model=child,
4854                config=config,
4855                stream_name=stream_name,
4856                **kwargs,
4857            )
4858            for child in model.partition_routers
4859        ]
4860
4861        # partition_field depends only on config/parameters, so it is evaluated once at build
4862        # time; the runtime component always receives a plain string.
4863        partition_field = InterpolatedString.create(
4864            model.partition_field, parameters=model.parameters or {}
4865        ).eval(config)
4866
4867        # Fail fast at build time when a built-in child router is statically known to emit a
4868        # partition field different from the union's. CustomPartitionRouter children are opaque
4869        # and can only be validated at runtime.
4870        for child_model in model.partition_routers:
4871            child_partition_fields: List[str] = []
4872            if isinstance(child_model, ListPartitionRouterModel):
4873                child_partition_fields.append(
4874                    InterpolatedString.create(
4875                        child_model.cursor_field, parameters=child_model.parameters or {}
4876                    ).eval(config)
4877                )
4878            elif isinstance(child_model, SubstreamPartitionRouterModel):
4879                for parent_stream_config in child_model.parent_stream_configs:
4880                    child_partition_fields.append(
4881                        InterpolatedString.create(
4882                            parent_stream_config.partition_field,
4883                            parameters=parent_stream_config.parameters
4884                            or child_model.parameters
4885                            or {},
4886                        ).eval(config)
4887                    )
4888            elif isinstance(child_model, UnionPartitionRouterModel):
4889                child_partition_fields.append(
4890                    InterpolatedString.create(
4891                        child_model.partition_field, parameters=child_model.parameters or {}
4892                    ).eval(config)
4893                )
4894            for child_partition_field in child_partition_fields:
4895                if child_partition_field != partition_field:
4896                    raise ValueError(
4897                        f"UnionPartitionRouter expects all child partition routers to emit the "
4898                        f"partition field '{partition_field}', but a "
4899                        f"{child_model.type} child emits '{child_partition_field}'."
4900                    )
4901
4902        # A union slice comes from exactly one child partition router, so request options
4903        # declared on children cannot be applied consistently to requests built from the
4904        # normalized union slices. Partition values should be consumed via interpolation
4905        # (e.g. stream_partition) instead. Note that this validation only covers built-in
4906        # router types; CustomPartitionRouter children are opaque, so any request options
4907        # they implement internally cannot be detected or rejected here.
4908        for router in partition_routers:
4909            if isinstance(router, SubstreamPartitionRouter):
4910                if any(
4911                    parent_config.request_option for parent_config in router.parent_stream_configs
4912                ):
4913                    raise ValueError("Request options are not supported for UnionPartitionRouter.")
4914            if isinstance(router, ListPartitionRouter) and router.request_option:
4915                raise ValueError("Request options are not supported for UnionPartitionRouter.")
4916
4917        return UnionPartitionRouter(
4918            partition_routers=partition_routers,
4919            partition_field=partition_field,
4920            parameters=model.parameters or {},
4921        )
4922
4923    def _ensure_query_properties_to_model(
4924        self, requester: Union[HttpRequesterModel, CustomRequesterModel]
4925    ) -> None:
4926        """
4927        For some reason, it seems like CustomRequesterModel request_parameters stays as dictionaries which means that
4928        the other conditions relying on it being QueryPropertiesModel instead of a dict fail. Here, we migrate them to
4929        proper model.
4930        """
4931        if not hasattr(requester, "request_parameters"):
4932            return
4933
4934        request_parameters = requester.request_parameters
4935        if request_parameters and isinstance(request_parameters, Dict):
4936            for request_parameter_key in request_parameters.keys():
4937                request_parameter = request_parameters[request_parameter_key]
4938                if (
4939                    isinstance(request_parameter, Dict)
4940                    and request_parameter.get("type") == "QueryProperties"
4941                ):
4942                    request_parameters[request_parameter_key] = QueryPropertiesModel.parse_obj(
4943                        request_parameter
4944                    )
4945
4946    def _get_catalog_defined_cursor_field(
4947        self, stream_name: str, allow_catalog_defined_cursor_field: bool
4948    ) -> Optional[CursorField]:
4949        if not allow_catalog_defined_cursor_field:
4950            return None
4951
4952        configured_stream = self._stream_name_to_configured_stream.get(stream_name)
4953
4954        # Depending on the operation is being performed, there may not be a configured stream yet. In this
4955        # case we return None which will then use the default cursor field defined on the cursor model.
4956        # We also treat cursor_field: [""] (list with empty string) as no cursor field, since this can
4957        # occur when the platform serializes "no cursor configured" streams incorrectly.
4958        if (
4959            not configured_stream
4960            or not configured_stream.cursor_field
4961            or not configured_stream.cursor_field[0]
4962        ):
4963            return None
4964        elif len(configured_stream.cursor_field) > 1:
4965            raise ValueError(
4966                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."
4967            )
4968        else:
4969            return CursorField(
4970                cursor_field_key=configured_stream.cursor_field[0],
4971                supports_catalog_defined_cursor_field=allow_catalog_defined_cursor_field,
4972            )