Airbyte API Client

0.53.0 · active · verified Thu Apr 09

The `airbyte-api` library is the official Python client SDK for interacting with the Airbyte API. It provides programmatic access to manage connections, sources, destinations, and perform data synchronization operations. The current version is 0.53.0 and it is actively maintained with frequent updates reflecting the underlying Airbyte platform and API.

Warnings

breaking The `airbyte-api` Python SDK is currently in beta. Breaking changes may occur in minor versions until the SDK reaches a stable v1.0.0 release. Always review the changelog before upgrading.
Fix: Refer to the official GitHub repository and changelog for release notes and migration guides during upgrades. Pin dependency versions to avoid unexpected breaks.
gotcha Incorrect `bearer_auth` or `server_url` can lead to authentication failures or 'not found' errors. Ensure your API key is correct and the `server_url` points to the correct Airbyte API endpoint (e.g., `https://api.airbyte.com/v1` for Airbyte Cloud, or your self-hosted instance's URL).
Fix: Verify your Airbyte API key and ensure the `server_url` matches your Airbyte deployment. For self-hosted, this often includes a port (e.g., `http://localhost:8000/api/v1`).
gotcha When making API calls, it's crucial to always check the `response.status_code` to ensure the call was successful and handle potential errors. Directly accessing `response.data` or similar attributes without checking status can lead to `AttributeError` if an error occurred and the expected data model is not present.
Fix: Always include `if response.status_code == 200:` (or other expected success codes) checks before processing the response body. Utilize `response.error_response` for detailed error messages when available.

Install

pip install airbyte-api Install `airbyte-api`

Imports

Airbyte
```
import airbyte_api
```
shared
```
from airbyte_api.models import shared
```

Quickstart

Initialize the Airbyte client with API key and server URL, then perform a simple API call to list workspaces. Remember to set `AIRBYTE_API_KEY` and `AIRBYTE_SERVER_URL` environment variables, or replace placeholders.

import airbyte_api
import os
from airbyte_api.models import shared

# Initialize the Airbyte client
client = airbyte_api.Airbyte(
    bearer_auth=os.environ.get('AIRBYTE_API_KEY', 'YOUR_API_KEY'),
    server_url=os.environ.get('AIRBYTE_SERVER_URL', 'https://api.airbyte.com/v1')
)

# Example: List workspaces
try:
    response = client.workspaces.list_workspaces()

    if response.status_code == 200:
        print("Successfully listed workspaces:")
        for workspace in response.list_workspaces_response.data:
            print(f"- ID: {workspace.workspace_id}, Name: {workspace.name}")
    else:
        print(f"Error listing workspaces: Status {response.status_code}")
        if response.error_response:
            print(f"Error details: {response.error_response.message}")
except Exception as e:
    print(f"An unexpected error occurred: {e}")

view raw JSON →