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, Mapping, Tuple 8 9from airbyte_cdk.sources.abstract_source import AbstractSource 10from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker 11 12 13class DeclarativeSource(AbstractSource): 14 """ 15 Base class for declarative Source. Concrete sources need to define the connection_checker to use 16 """ 17 18 @property 19 @abstractmethod 20 def connection_checker(self) -> ConnectionChecker: 21 """Returns the ConnectionChecker to use for the `check` operation""" 22 23 def check_connection( 24 self, logger: logging.Logger, config: Mapping[str, Any] 25 ) -> Tuple[bool, Any]: 26 """ 27 :param logger: The source logger 28 :param config: The user-provided configuration as specified by the source's spec. 29 This usually contains information required to check connection e.g. tokens, secrets and keys etc. 30 :return: A tuple of (boolean, error). If boolean is true, then the connection check is successful 31 and we can connect to the underlying data source using the provided configuration. 32 Otherwise, the input config cannot be used to connect to the underlying data source, 33 and the "error" object should describe what went wrong. 34 The error object will be cast to string to display the problem to the user. 35 """ 36 return self.connection_checker.check_connection(self, logger, config)
class
DeclarativeSource(airbyte_cdk.connector.DefaultConnectorMixin, airbyte_cdk.sources.source.BaseSource[typing.Mapping[str, typing.Any], typing.List[airbyte_cdk.models.airbyte_protocol.AirbyteStateMessage], airbyte_protocol_dataclasses.models.airbyte_protocol.ConfiguredAirbyteCatalog], abc.ABC):
14class DeclarativeSource(AbstractSource): 15 """ 16 Base class for declarative Source. Concrete sources need to define the connection_checker to use 17 """ 18 19 @property 20 @abstractmethod 21 def connection_checker(self) -> ConnectionChecker: 22 """Returns the ConnectionChecker to use for the `check` operation""" 23 24 def check_connection( 25 self, logger: logging.Logger, config: Mapping[str, Any] 26 ) -> Tuple[bool, Any]: 27 """ 28 :param logger: The source logger 29 :param config: The user-provided configuration as specified by the source's spec. 30 This usually contains information required to check connection e.g. tokens, secrets and keys etc. 31 :return: A tuple of (boolean, error). If boolean is true, then the connection check is successful 32 and we can connect to the underlying data source using the provided configuration. 33 Otherwise, the input config cannot be used to connect to the underlying data source, 34 and the "error" object should describe what went wrong. 35 The error object will be cast to string to display the problem to the user. 36 """ 37 return self.connection_checker.check_connection(self, logger, config)
Base class for declarative Source. Concrete sources need to define the connection_checker to use
connection_checker: airbyte_cdk.sources.declarative.checks.ConnectionChecker
19 @property 20 @abstractmethod 21 def connection_checker(self) -> ConnectionChecker: 22 """Returns the ConnectionChecker to use for the `check` operation"""
Returns the ConnectionChecker to use for the check operation
def
check_connection( self, logger: logging.Logger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
24 def check_connection( 25 self, logger: logging.Logger, config: Mapping[str, Any] 26 ) -> Tuple[bool, Any]: 27 """ 28 :param logger: The source logger 29 :param config: The user-provided configuration as specified by the source's spec. 30 This usually contains information required to check connection e.g. tokens, secrets and keys etc. 31 :return: A tuple of (boolean, error). If boolean is true, then the connection check is successful 32 and we can connect to the underlying data source using the provided configuration. 33 Otherwise, the input config cannot be used to connect to the underlying data source, 34 and the "error" object should describe what went wrong. 35 The error object will be cast to string to display the problem to the user. 36 """ 37 return self.connection_checker.check_connection(self, logger, config)
Parameters
- logger: The source logger
- config: The user-provided configuration as specified by the source's spec. This usually contains information required to check connection e.g. tokens, secrets and keys etc.
Returns
A tuple of (boolean, error). If boolean is true, then the connection check is successful and we can connect to the underlying data source using the provided configuration. Otherwise, the input config cannot be used to connect to the underlying data source, and the "error" object should describe what went wrong. The error object will be cast to string to display the problem to the user.