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