Kaustav Ghosh
10/21/2024, 9:13 AMuser
10/21/2024, 9:13 AMcheck
function in your connector. This function verifies that the input configuration supplied by the user can be used to connect to the underlying data source. Here is an example of how to implement the check
function: python def check_connection(self, logger, config) -> Tuple[bool, any]: first_stream = next(iter(self.streams(config))) stream_slice = next(iter(first_stream.stream_slices(sync_mode=SyncMode.full_refresh))) try: read_stream = first_stream.read_records(sync_mode=SyncMode.full_refresh, stream_slice=stream_slice) first_record = None while not first_record: first_record = next(read_stream) if isinstance(first_record, AirbyteMessage): if first_record.type == "RECORD": first_record = first_record.record return True, None else: first_record = None return True, None except Exception as e: return False, f"Unable to connect to the API with the provided credentials - {str(e)}"
(Source: [Airbyte Documentation](https://docs.airbyte.com/connector-development/tutorials/custom-python-connector/check-and-error-handling)) 2. Run the check
operation using the following command: bash poetry run source-survey-monkey-demo check --config secrets/config.json
3. If the command succeeds, it means your connector is able to connect to the data source. If it fails, you should see an error message indicating what went wrong. 4. You can also test your connector locally with Python and a test file. This involves defining needed mock data stored in a JSON file and running a series of commands to test the spec
, check
, discover
, and write
functions. Here is an example of how to do this: bash source .venv/bin/activate python -m pip install --upgrade pip pip install -r requirements.txt python main.py spec python main.py check --config integration_tests/config.json python main.py discover --config integration_tests/config.json cat integration_tests/messages.jsonl| python main.py write --config integration_tests/config.json --catalog integration_tests/configured_catalog.json
(Source: [Airbyte Tutorials](https://airbyte.com/tutorials/creating-duckdb-destination-with-python)) 5. Finally, make sure your connector passes airbyte-ci connectors test
tests. You can find a guide on how to run them [here](https://docs.airbyte.com/connector-development/testing-connectors/). (Source: [Airbyte Documentation](https://docs.airbyte.com/contributing-to-airbyte/submit-new-connector#open-a-pull-request))