8bb41282c5
* Revert "source-github: move known error handling to GithubAvailabilityStrategy (#19978)" This reverts commitf97db17ccc. * Revert "🐛 Python CDK: fix `StopIteration` error for `check_availability` (#20429)" This reverts commit4e9b014277. * Revert "CDK: `AbstractSource.read()` skips syncing stream if its unavailable (add `AvailabilityStrategy` concept) (#19977)" This reverts commit55a32886a3. * Restore changelog entries * bump CDK version * Bump Github version * Re-add removed dependencies * auto-bump connector version Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
34 lines
1.4 KiB
Python
34 lines
1.4 KiB
Python
#
|
|
# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
from abc import abstractmethod
|
|
from typing import Tuple
|
|
|
|
from airbyte_cdk.sources.abstract_source import AbstractSource
|
|
from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker
|
|
|
|
|
|
class DeclarativeSource(AbstractSource):
|
|
"""
|
|
Base class for declarative Source. Concrete sources need to define the connection_checker to use
|
|
"""
|
|
|
|
@property
|
|
@abstractmethod
|
|
def connection_checker(self) -> ConnectionChecker:
|
|
"""Returns the ConnectioChecker to use for the `check` operation"""
|
|
|
|
def check_connection(self, logger, config) -> Tuple[bool, any]:
|
|
"""
|
|
:param logger: The source logger
|
|
:param 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.
|
|
:return: 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.
|
|
"""
|
|
return self.connection_checker.check_connection(self, logger, config)
|