airbyte_cdk.legacy.sources.declarative.manifest_declarative_source

  1#
  2# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  3#
  4
  5import json
  6import logging
  7import pkgutil
  8from copy import deepcopy
  9from importlib import metadata
 10from types import ModuleType
 11from typing import Any, Dict, Iterator, List, Mapping, Optional, Set, Union
 12
 13import orjson
 14from jsonschema.exceptions import ValidationError
 15from packaging.version import InvalidVersion, Version
 16
 17from airbyte_cdk.config_observation import create_connector_config_control_message
 18from airbyte_cdk.connector_builder.models import (
 19    LogMessage as ConnectorBuilderLogMessage,
 20)
 21from airbyte_cdk.legacy.sources.declarative.declarative_source import DeclarativeSource
 22from airbyte_cdk.manifest_migrations.migration_handler import (
 23    ManifestMigrationHandler,
 24)
 25from airbyte_cdk.models import (
 26    AirbyteConnectionStatus,
 27    AirbyteMessage,
 28    AirbyteStateMessage,
 29    ConfiguredAirbyteCatalog,
 30    ConnectorSpecification,
 31    FailureType,
 32)
 33from airbyte_cdk.models.airbyte_protocol_serializers import AirbyteMessageSerializer
 34from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING
 35from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
 36from airbyte_cdk.sources.declarative.concurrent_declarative_source import validate_manifest
 37from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean
 38from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 39    ConditionalStreams as ConditionalStreamsModel,
 40)
 41from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 42    DeclarativeStream as DeclarativeStreamModel,
 43)
 44from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 45    Spec as SpecModel,
 46)
 47from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
 48    StateDelegatingStream as StateDelegatingStreamModel,
 49)
 50from airbyte_cdk.sources.declarative.parsers.custom_code_compiler import (
 51    get_registered_components_module,
 52)
 53from airbyte_cdk.sources.declarative.parsers.manifest_component_transformer import (
 54    ManifestComponentTransformer,
 55)
 56from airbyte_cdk.sources.declarative.parsers.manifest_normalizer import (
 57    ManifestNormalizer,
 58)
 59from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import (
 60    ManifestReferenceResolver,
 61)
 62from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
 63    ModelToComponentFactory,
 64)
 65from airbyte_cdk.sources.declarative.resolvers import COMPONENTS_RESOLVER_TYPE_MAPPING
 66from airbyte_cdk.sources.declarative.spec.spec import Spec
 67from airbyte_cdk.sources.message import MessageRepository
 68from airbyte_cdk.sources.streams.concurrent.abstract_stream import AbstractStream
 69from airbyte_cdk.sources.streams.core import Stream
 70from airbyte_cdk.sources.types import Config, ConnectionDefinition
 71from airbyte_cdk.sources.utils.slice_logger import (
 72    AlwaysLogSliceLogger,
 73    DebugSliceLogger,
 74    SliceLogger,
 75)
 76from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 77from airbyte_cdk.utils.yaml_loader import safe_load_yaml
 78
 79
 80def _get_declarative_component_schema() -> Dict[str, Any]:
 81    try:
 82        raw_component_schema = pkgutil.get_data(
 83            "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml"
 84        )
 85        if raw_component_schema is not None:
 86            declarative_component_schema = safe_load_yaml(raw_component_schema)
 87            return declarative_component_schema  # type: ignore
 88        else:
 89            raise RuntimeError(
 90                "Failed to read manifest component json schema required for deduplication"
 91            )
 92    except FileNotFoundError as e:
 93        raise FileNotFoundError(
 94            f"Failed to read manifest component json schema required for deduplication: {e}"
 95        )
 96
 97
 98class ManifestDeclarativeSource(DeclarativeSource):
 99    """Declarative source defined by a manifest of low-code components that define source connector behavior"""
100
101    def __init__(
102        self,
103        source_config: ConnectionDefinition,
104        *,
105        config: Mapping[str, Any] | None = None,
106        debug: bool = False,
107        emit_connector_builder_messages: bool = False,
108        component_factory: Optional[ModelToComponentFactory] = None,
109        migrate_manifest: Optional[bool] = False,
110        normalize_manifest: Optional[bool] = False,
111        config_path: Optional[str] = None,
112    ) -> None:
113        """
114        Args:
115            config: The provided config dict.
116            source_config: The manifest of low-code components that describe the source connector.
117            debug: True if debug mode is enabled.
118            emit_connector_builder_messages: True if messages should be emitted to the connector builder.
119            component_factory: optional factory if ModelToComponentFactory's default behavior needs to be tweaked.
120            normalize_manifest: Optional flag to indicate if the manifest should be normalized.
121            config_path: Optional path to the config file.
122        """
123        self.logger = logging.getLogger(f"airbyte.{self.name}")
124        self._should_normalize = normalize_manifest
125        self._should_migrate = migrate_manifest
126        self._declarative_component_schema = _get_declarative_component_schema()
127        # If custom components are needed, locate and/or register them.
128        self.components_module: ModuleType | None = get_registered_components_module(config=config)
129        # set additional attributes
130        self._debug = debug
131        self._emit_connector_builder_messages = emit_connector_builder_messages
132        self._constructor = (
133            component_factory
134            if component_factory
135            else ModelToComponentFactory(
136                emit_connector_builder_messages=emit_connector_builder_messages,
137                max_concurrent_async_job_count=source_config.get("max_concurrent_async_job_count"),
138            )
139        )
140        self._message_repository = self._constructor.get_message_repository()
141        self._slice_logger: SliceLogger = (
142            AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
143        )
144
145        # resolve all components in the manifest
146        self._source_config = self._pre_process_manifest(dict(source_config))
147        # validate resolved manifest against the declarative component schema
148        self._validate_source()
149        # apply additional post-processing to the manifest
150        self._post_process_manifest()
151
152        spec: Optional[Mapping[str, Any]] = self._source_config.get("spec")
153        self._spec_component: Optional[Spec] = (
154            self._constructor.create_component(SpecModel, spec, dict()) if spec else None
155        )
156        self._config = self._migrate_and_transform_config(config_path, config) or {}
157
158    @property
159    def resolved_manifest(self) -> Mapping[str, Any]:
160        """
161        Returns the resolved manifest configuration for the source.
162
163        This property provides access to the internal source configuration as a mapping,
164        which contains all settings and parameters required to define the source's behavior.
165
166        Returns:
167            Mapping[str, Any]: The resolved source configuration manifest.
168        """
169        return self._source_config
170
171    def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
172        """
173        Preprocesses the provided manifest dictionary by resolving any manifest references.
174
175        This method modifies the input manifest in place, resolving references using the
176        ManifestReferenceResolver to ensure all references within the manifest are properly handled.
177
178        Args:
179            manifest (Dict[str, Any]): The manifest dictionary to preprocess and resolve references in.
180
181        Returns:
182            None
183        """
184        # For ease of use we don't require the type to be specified at the top level manifest, but it should be included during processing
185        manifest = self._fix_source_type(manifest)
186        # Resolve references in the manifest
187        resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest)
188        # Propagate types and parameters throughout the manifest
189        propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters(
190            "", resolved_manifest, {}
191        )
192
193        return propagated_manifest
194
195    def _post_process_manifest(self) -> None:
196        """
197        Post-processes the manifest after validation.
198        This method is responsible for any additional modifications or transformations needed
199        after the manifest has been validated and before it is used in the source.
200        """
201        # apply manifest migration, if required
202        self._migrate_manifest()
203        # apply manifest normalization, if required
204        self._normalize_manifest()
205
206    def _normalize_manifest(self) -> None:
207        """
208        This method is used to normalize the manifest. It should be called after the manifest has been validated.
209
210        Connector Builder UI rendering requires the manifest to be in a specific format.
211         - references have been resolved
212         - the commonly used definitions are extracted to the `definitions.linked.*`
213        """
214        if self._should_normalize:
215            normalizer = ManifestNormalizer(self._source_config, self._declarative_component_schema)
216            self._source_config = normalizer.normalize()
217
218    def _migrate_and_transform_config(
219        self,
220        config_path: Optional[str],
221        config: Optional[Config],
222    ) -> Optional[Config]:
223        if not config:
224            return None
225        if not self._spec_component:
226            return config
227        mutable_config = dict(config)
228        self._spec_component.migrate_config(mutable_config)
229        if mutable_config != config:
230            if config_path:
231                with open(config_path, "w") as f:
232                    json.dump(mutable_config, f)
233            self.message_repository.emit_message(
234                create_connector_config_control_message(mutable_config)
235            )
236            # We have no mechanism for consuming the queue, so we print the messages to stdout
237            for message in self.message_repository.consume_queue():
238                print(orjson.dumps(AirbyteMessageSerializer.dump(message)).decode())
239        self._spec_component.transform_config(mutable_config)
240        return mutable_config
241
242    def configure(self, config: Mapping[str, Any], temp_dir: str) -> Mapping[str, Any]:
243        config = self._config or config
244        return super().configure(config, temp_dir)
245
246    def _migrate_manifest(self) -> None:
247        """
248        This method is used to migrate the manifest. It should be called after the manifest has been validated.
249        The migration is done in place, so the original manifest is modified.
250
251        The original manifest is returned if any error occurs during migration.
252        """
253        if self._should_migrate:
254            manifest_migrator = ManifestMigrationHandler(self._source_config)
255            self._source_config = manifest_migrator.apply_migrations()
256            # validate migrated manifest against the declarative component schema
257            self._validate_source()
258
259    def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
260        """
261        Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest.
262        """
263        if "type" not in manifest:
264            manifest["type"] = "DeclarativeSource"
265
266        return manifest
267
268    @property
269    def message_repository(self) -> MessageRepository:
270        return self._message_repository
271
272    @property
273    def dynamic_streams(self) -> List[Dict[str, Any]]:
274        return self._dynamic_stream_configs(
275            manifest=self._source_config,
276            config=self._config,
277            with_dynamic_stream_name=True,
278        )
279
280    def deprecation_warnings(self) -> List[ConnectorBuilderLogMessage]:
281        return self._constructor.get_model_deprecations()
282
283    @property
284    def connection_checker(self) -> ConnectionChecker:
285        check = self._source_config["check"]
286        if "type" not in check:
287            check["type"] = "CheckStream"
288        check_stream = self._constructor.create_component(
289            COMPONENTS_CHECKER_TYPE_MAPPING[check["type"]],
290            check,
291            dict(),
292            emit_connector_builder_messages=self._emit_connector_builder_messages,
293        )
294        if isinstance(check_stream, ConnectionChecker):
295            return check_stream
296        else:
297            raise ValueError(
298                f"Expected to generate a ConnectionChecker component, but received {check_stream.__class__}"
299            )
300
301    def streams(self, config: Mapping[str, Any]) -> List[Union[Stream, AbstractStream]]:  # type: ignore  # we are migrating away from the AbstractSource and are expecting that this will only be called by ConcurrentDeclarativeSource or the Connector Builder
302        """
303        As a migration step, this method will return both legacy stream (Stream) and concurrent stream (AbstractStream).
304        Once the migration is done, we can probably have this method throw "not implemented" as we figure out how to
305        fully decouple this from the AbstractSource.
306        """
307        if self._spec_component:
308            self._spec_component.validate_config(config)
309
310        self._emit_manifest_debug_message(
311            extra_args={
312                "source_name": self.name,
313                "parsed_config": json.dumps(self._source_config),
314            }
315        )
316
317        api_budget_model = self._source_config.get("api_budget")
318        if api_budget_model:
319            self._constructor.set_api_budget(api_budget_model, config)
320
321        stream_configs = (
322            self._stream_configs(self._source_config, config=config) + self.dynamic_streams
323        )
324
325        source_streams = [
326            self._constructor.create_component(
327                (
328                    StateDelegatingStreamModel
329                    if stream_config.get("type") == StateDelegatingStreamModel.__name__
330                    else DeclarativeStreamModel
331                ),
332                stream_config,
333                config,
334                emit_connector_builder_messages=self._emit_connector_builder_messages,
335            )
336            for stream_config in self._initialize_cache_for_parent_streams(deepcopy(stream_configs))
337        ]
338        return source_streams
339
340    @staticmethod
341    def _initialize_cache_for_parent_streams(
342        stream_configs: List[Dict[str, Any]],
343    ) -> List[Dict[str, Any]]:
344        """Enable caching for parent streams unless explicitly disabled.
345
346        Caching is enabled by default for parent streams to optimize performance when the same
347        parent data is needed by multiple child streams. However, explicit `use_cache: false`
348        settings are respected for streams that cannot use caching (e.g., scroll-based pagination
349        APIs where caching causes duplicate records).
350        """
351        parent_streams = set()
352
353        def _set_cache_if_not_disabled(requester: Dict[str, Any]) -> None:
354            """Set use_cache to True only if not explicitly disabled."""
355            if requester.get("use_cache") is not False:
356                requester["use_cache"] = True
357
358        def update_with_cache_parent_configs(
359            parent_configs: list[dict[str, Any]],
360        ) -> None:
361            for parent_config in parent_configs:
362                parent_streams.add(parent_config["stream"]["name"])
363                if parent_config["stream"]["type"] == "StateDelegatingStream":
364                    _set_cache_if_not_disabled(
365                        parent_config["stream"]["full_refresh_stream"]["retriever"]["requester"]
366                    )
367                    _set_cache_if_not_disabled(
368                        parent_config["stream"]["incremental_stream"]["retriever"]["requester"]
369                    )
370                else:
371                    _set_cache_if_not_disabled(parent_config["stream"]["retriever"]["requester"])
372
373        for stream_config in stream_configs:
374            if stream_config.get("incremental_sync", {}).get("parent_stream"):
375                parent_streams.add(stream_config["incremental_sync"]["parent_stream"]["name"])
376                _set_cache_if_not_disabled(
377                    stream_config["incremental_sync"]["parent_stream"]["retriever"]["requester"]
378                )
379
380            elif stream_config.get("retriever", {}).get("partition_router", {}):
381                partition_router = stream_config["retriever"]["partition_router"]
382
383                if isinstance(partition_router, dict) and partition_router.get(
384                    "parent_stream_configs"
385                ):
386                    update_with_cache_parent_configs(partition_router["parent_stream_configs"])
387                elif isinstance(partition_router, list):
388                    for router in partition_router:
389                        if router.get("parent_stream_configs"):
390                            update_with_cache_parent_configs(router["parent_stream_configs"])
391
392        for stream_config in stream_configs:
393            if stream_config["name"] in parent_streams:
394                if stream_config["type"] == "StateDelegatingStream":
395                    _set_cache_if_not_disabled(
396                        stream_config["full_refresh_stream"]["retriever"]["requester"]
397                    )
398                    _set_cache_if_not_disabled(
399                        stream_config["incremental_stream"]["retriever"]["requester"]
400                    )
401                else:
402                    _set_cache_if_not_disabled(stream_config["retriever"]["requester"])
403        return stream_configs
404
405    def spec(self, logger: logging.Logger) -> ConnectorSpecification:
406        """
407        Returns the connector specification (spec) as defined in the Airbyte Protocol. The spec is an object describing the possible
408        configurations (e.g: username and password) which can be configured when running this connector. For low-code connectors, this
409        will first attempt to load the spec from the manifest's spec block, otherwise it will load it from "spec.yaml" or "spec.json"
410        in the project root.
411        """
412        self._configure_logger_level(logger)
413        self._emit_manifest_debug_message(
414            extra_args={
415                "source_name": self.name,
416                "parsed_config": json.dumps(self._source_config),
417            }
418        )
419
420        return (
421            self._spec_component.generate_spec() if self._spec_component else super().spec(logger)
422        )
423
424    def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteConnectionStatus:
425        self._configure_logger_level(logger)
426        return super().check(logger, config)
427
428    def read(
429        self,
430        logger: logging.Logger,
431        config: Mapping[str, Any],
432        catalog: ConfiguredAirbyteCatalog,
433        state: Optional[List[AirbyteStateMessage]] = None,
434    ) -> Iterator[AirbyteMessage]:
435        self._configure_logger_level(logger)
436        yield from super().read(logger, config, catalog, state)
437
438    def _configure_logger_level(self, logger: logging.Logger) -> None:
439        """
440        Set the log level to logging.DEBUG if debug mode is enabled
441        """
442        if self._debug:
443            logger.setLevel(logging.DEBUG)
444
445    def _validate_source(self) -> None:
446        """
447        Validates the connector manifest against the declarative component schema
448        """
449
450        validate_manifest(self._source_config, self._declarative_component_schema)
451
452        cdk_version_str = metadata.version("airbyte_cdk")
453        cdk_version = self._parse_version(cdk_version_str, "airbyte-cdk")
454        manifest_version_str = self._source_config.get("version")
455        if manifest_version_str is None:
456            raise RuntimeError(
457                "Manifest version is not defined in the manifest. This is unexpected since it should be a required field. Please contact support."
458            )
459        manifest_version = self._parse_version(manifest_version_str, "manifest")
460
461        if (cdk_version.major, cdk_version.minor, cdk_version.micro) == (0, 0, 0):
462            # Skipping version compatibility check on unreleased dev branch
463            pass
464        elif (cdk_version.major, cdk_version.minor) < (
465            manifest_version.major,
466            manifest_version.minor,
467        ):
468            raise ValidationError(
469                f"The manifest version {manifest_version!s} is greater than the airbyte-cdk package version ({cdk_version!s}). Your "
470                f"manifest may contain features that are not in the current CDK version."
471            )
472        elif (manifest_version.major, manifest_version.minor) < (0, 29):
473            raise ValidationError(
474                f"The low-code framework was promoted to Beta in airbyte-cdk version 0.29.0 and contains many breaking changes to the "
475                f"language. The manifest version {manifest_version!s} is incompatible with the airbyte-cdk package version "
476                f"{cdk_version!s} which contains these breaking changes."
477            )
478
479    @staticmethod
480    def _parse_version(
481        version: str,
482        version_type: str,
483    ) -> Version:
484        """Takes a semantic version represented as a string and splits it into a tuple.
485
486        The fourth part (prerelease) is not returned in the tuple.
487
488        Returns:
489            Version: the parsed version object
490        """
491        try:
492            parsed_version = Version(version)
493        except InvalidVersion as ex:
494            raise ValidationError(
495                f"The {version_type} version '{version}' is not a valid version format."
496            ) from ex
497        else:
498            # No exception
499            return parsed_version
500
501    def _stream_configs(
502        self, manifest: Mapping[str, Any], config: Mapping[str, Any]
503    ) -> List[Dict[str, Any]]:
504        # This has a warning flag for static, but after we finish part 4 we'll replace manifest with self._source_config
505        stream_configs = []
506        for current_stream_config in manifest.get("streams", []):
507            if (
508                "type" in current_stream_config
509                and current_stream_config["type"] == "ConditionalStreams"
510            ):
511                interpolated_boolean = InterpolatedBoolean(
512                    condition=current_stream_config.get("condition"),
513                    parameters={},
514                )
515
516                if interpolated_boolean.eval(config=config):
517                    stream_configs.extend(current_stream_config.get("streams", []))
518            else:
519                if "type" not in current_stream_config:
520                    current_stream_config["type"] = "DeclarativeStream"
521                stream_configs.append(current_stream_config)
522        return stream_configs
523
524    def _dynamic_stream_configs(
525        self,
526        manifest: Mapping[str, Any],
527        config: Mapping[str, Any],
528        with_dynamic_stream_name: Optional[bool] = None,
529    ) -> List[Dict[str, Any]]:
530        dynamic_stream_definitions: List[Dict[str, Any]] = manifest.get("dynamic_streams", [])
531        dynamic_stream_configs: List[Dict[str, Any]] = []
532        seen_dynamic_streams: Set[str] = set()
533
534        for dynamic_definition_index, dynamic_definition in enumerate(dynamic_stream_definitions):
535            components_resolver_config = dynamic_definition["components_resolver"]
536
537            if not components_resolver_config:
538                raise ValueError(
539                    f"Missing 'components_resolver' in dynamic definition: {dynamic_definition}"
540                )
541
542            resolver_type = components_resolver_config.get("type")
543            if not resolver_type:
544                raise ValueError(
545                    f"Missing 'type' in components resolver configuration: {components_resolver_config}"
546                )
547
548            if resolver_type not in COMPONENTS_RESOLVER_TYPE_MAPPING:
549                raise ValueError(
550                    f"Invalid components resolver type '{resolver_type}'. "
551                    f"Expected one of {list(COMPONENTS_RESOLVER_TYPE_MAPPING.keys())}."
552                )
553
554            # AsyncRetriever has no `requester`, and a cached polling request replays a stale status
555            if "requester" in components_resolver_config.get("retriever", {}):
556                components_resolver_config["retriever"]["requester"]["use_cache"] = True
557
558            # Create a resolver for dynamic components based on type
559            if resolver_type == "HttpComponentsResolver":
560                components_resolver = self._constructor.create_component(
561                    model_type=COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
562                    component_definition=components_resolver_config,
563                    config=config,
564                    stream_name=dynamic_definition.get("name"),
565                )
566            else:
567                components_resolver = self._constructor.create_component(
568                    model_type=COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
569                    component_definition=components_resolver_config,
570                    config=config,
571                )
572
573            stream_template_config = dynamic_definition["stream_template"]
574
575            for dynamic_stream in components_resolver.resolve_components(
576                stream_template_config=stream_template_config
577            ):
578                # Get the use_parent_parameters configuration from the dynamic definition
579                # Default to True for backward compatibility, since connectors were already using it by default when this param was added
580                use_parent_parameters = dynamic_definition.get("use_parent_parameters", True)
581
582                dynamic_stream = {
583                    **ManifestComponentTransformer().propagate_types_and_parameters(
584                        "", dynamic_stream, {}, use_parent_parameters=use_parent_parameters
585                    )
586                }
587
588                if "type" not in dynamic_stream:
589                    dynamic_stream["type"] = "DeclarativeStream"
590
591                # Ensure that each stream is created with a unique name
592                name = dynamic_stream.get("name")
593
594                if with_dynamic_stream_name:
595                    dynamic_stream["dynamic_stream_name"] = dynamic_definition.get(
596                        "name", f"dynamic_stream_{dynamic_definition_index}"
597                    )
598
599                if not isinstance(name, str):
600                    raise ValueError(
601                        f"Expected stream name {name} to be a string, got {type(name)}."
602                    )
603
604                if name in seen_dynamic_streams:
605                    error_message = f"Dynamic streams list contains a duplicate name: {name}. Please contact Airbyte Support."
606                    failure_type = FailureType.system_error
607
608                    if resolver_type == "ConfigComponentsResolver":
609                        error_message = f"Dynamic streams list contains a duplicate name: {name}. Please check your configuration."
610                        failure_type = FailureType.config_error
611
612                    raise AirbyteTracedException(
613                        message=error_message,
614                        internal_message=error_message,
615                        failure_type=failure_type,
616                    )
617
618                seen_dynamic_streams.add(name)
619                dynamic_stream_configs.append(dynamic_stream)
620
621        return dynamic_stream_configs
622
623    def _emit_manifest_debug_message(self, extra_args: dict[str, Any]) -> None:
624        self.logger.debug("declarative source created from manifest", extra=extra_args)