0e70491e0f
* feat(67): add support for 'spec' using Python HTTP API source template and stripe as an example * chore(67): add sample state and config * feat(67): add check functionality for paystack source by fetching first customer * feat(67): add support for discover and read customer stream * feat(67): add paystack source connector to UI * feat(67): update source definitions to use 0.1.0 * Hacktoberfest 67 paystack source (#1) * feat(67): add support for 'spec' using Python HTTP API source template and stripe as an example * chore(67): add sample state and config * feat(67): add check functionality for paystack source by fetching first customer * feat(67): add support for discover and read customer stream * feat(67): add paystack source connector to UI * feat(67): update source definitions to use 0.1.0 Co-authored-by: Foluso Ogunlana <foluso_ogunlana@stearsng.com> * feat(67): update stream state cursor field to be integer and to match API record field name * chore(67): add unit tests for source and streams * chore(67): store formatted date time in state to match type of catalog * chore(67): add configuration for acceptance integration tests * docs(67): update docs and summary with paystack * chore(67): add essential schemas to be catalogued for new streams * feat(67): add support for critical streams - transactions subscriptions transfers refunds settlements * docs(67): update image and bootstrap * chore(67): update builds.md to include paystack badge * docs(67): add changelog and source definition JSON file * docs(67): add paystack to integrations readme * chore(67): update check_connection to airbyte standard * refactor to simplify streams and remove constants file * fix(67): correct "null, null" values in schemas * chore(67): update file formatting with gradle format Co-authored-by: Foluso <5675998+foogunlana@users.noreply.github.com>
56 lines
2.2 KiB
Python
56 lines
2.2 KiB
Python
#
|
|
# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
|
|
#
|
|
|
|
from typing import Any, List, Mapping, Tuple
|
|
|
|
from airbyte_cdk.models import SyncMode
|
|
from airbyte_cdk.sources import AbstractSource
|
|
from airbyte_cdk.sources.streams import Stream
|
|
from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
|
|
from source_paystack.streams import Customers, Disputes, Invoices, Refunds, Settlements, Subscriptions, Transactions, Transfers
|
|
|
|
|
|
class SourcePaystack(AbstractSource):
|
|
def check_connection(self, logger, config) -> Tuple[bool, any]:
|
|
"""
|
|
Check connection by fetching customers
|
|
|
|
:param config: the user-input config object conforming to the connector's spec.json
|
|
:param logger: logger object
|
|
:return Tuple[bool, any]: (True, None) if the input config can be used to connect to the API successfully, (False, error) otherwise.
|
|
"""
|
|
try:
|
|
authenticator = TokenAuthenticator(token=config["secret_key"])
|
|
stream = Customers(authenticator=authenticator, start_date=config["start_date"])
|
|
records = stream.read_records(sync_mode=SyncMode.full_refresh)
|
|
next(records)
|
|
return True, None
|
|
|
|
except StopIteration:
|
|
# there are no records, but connection was fine
|
|
return True, None
|
|
|
|
except Exception as e:
|
|
return False, repr(e)
|
|
|
|
def streams(self, config: Mapping[str, Any]) -> List[Stream]:
|
|
"""
|
|
Returns list of streams output by the Paystack source connector
|
|
|
|
:param config: A Mapping of the user input configuration as defined in the connector spec.
|
|
"""
|
|
authenticator = TokenAuthenticator(config["secret_key"])
|
|
args = {"authenticator": authenticator, "start_date": config["start_date"]}
|
|
incremental_args = {**args, "lookback_window_days": config.get("lookback_window_days")}
|
|
return [
|
|
Customers(**incremental_args),
|
|
Disputes(**incremental_args),
|
|
Invoices(**incremental_args),
|
|
Refunds(**incremental_args),
|
|
Settlements(**incremental_args),
|
|
Subscriptions(**incremental_args),
|
|
Transactions(**incremental_args),
|
|
Transfers(**incremental_args),
|
|
]
|