airbyte_cdk.sources.declarative.declarative_source

 1#
 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 3#
 4
 5import logging
 6from abc import abstractmethod
 7from typing import Any, List, Mapping, Tuple
 8
 9from airbyte_cdk.connector_builder.models import (
10    LogMessage as ConnectorBuilderLogMessage,
11)
12from airbyte_cdk.sources.abstract_source import AbstractSource
13from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
14
15
16class DeclarativeSource(AbstractSource):
17    """
18    Base class for declarative Source. Concrete sources need to define the connection_checker to use
19    """
20
21    @property
22    @abstractmethod
23    def connection_checker(self) -> ConnectionChecker:
24        """Returns the ConnectionChecker to use for the `check` operation"""
25
26    def check_connection(
27        self, logger: logging.Logger, config: Mapping[str, Any]
28    ) -> Tuple[bool, Any]:
29        """
30        :param logger: The source logger
31        :param config: The user-provided configuration as specified by the source's spec.
32          This usually contains information required to check connection e.g. tokens, secrets and keys etc.
33        :return: A tuple of (boolean, error). If boolean is true, then the connection check is successful
34          and we can connect to the underlying data source using the provided configuration.
35          Otherwise, the input config cannot be used to connect to the underlying data source,
36          and the "error" object should describe what went wrong.
37          The error object will be cast to string to display the problem to the user.
38        """
39        return self.connection_checker.check_connection(self, logger, config)
40
41    def deprecation_warnings(self) -> List[ConnectorBuilderLogMessage]:
42        """
43        Returns a list of deprecation warnings for the source.
44        """
45        return []