Article

Manage Airbyte Programmatically: A Guide to the API, Terraform, and PyAirbyte

•

November 10, 2024

•

15 min read

Summarize with ChatGPT

While Airbyte has an intuitive interface where you can manage your various syncs, using the UI is not desirable in certain scenarios—especially in automated or large-scale environments. Luckily, Airbyte offers several options for managing syncs programmatically, so you can automate tasks without needing manual intervention.

Programmatically managing Airbyte is ideal for those who want to include it as part of an orchestrated data pipeline, integrating it into CI/CD pipelines, incorporating it as part of Infrastructure as Code (IaC), or version-controlling configurations. Airbyte’s flexible tools make it easy to handle data processes with minimal oversight.

There are three main ways to programmatically manage Airbyte:

The Airbyte API
PyAirbyte
Terraform

In this article, I will share more about each feature and how you can use it to simplify your data processes.

Why manage Airbyte programmatically?

Adding another tool to your already-existing code base can feel overwhelming. Modern data stacks sometimes feel like a bowl of spaghetti with all of the different tools working together, each serving its own purpose. Visiting multiple different UIs to paint a clear picture of how everything is working can feel like a full-time job.

When you manage tools programmatically, you manage them directly from the code that you are already interacting with daily. No additional UI is added to your data stack! Everything you need to monitor data quality and synchronicity between tools is in one central location—your code.

Programmatically managing code in one central location helps to improve efficiency through automatic updates. Again, no visits to the UI and manual actions are necessary. This way of working reduces human errors since changes and updates are all happening through your code. With all changes tracked in code, version control makes it easier to troubleshoot issues and quickly roll back to previous states when needed.

1. The Airbyte API

The Airbyte API allows you to make requests to access the most popular Airbyte objects, retrieving, deleting, and updating them programmatically. It gives you greater flexibility in using Airbyte in your software and data products, providing you with similar functionality that you get in the UI.

With the Airbyte API, you can trigger a sync within your data pipeline using an orchestration tool like Dagster or Prefect. This means your data will ingest and then your transformations will run on this newly synced data, all without a single user touching the UI.

The API also allows you to create new sources, destinations, and connections directly within your code. This means you could create a process that automatically generates a new source and connection every time a specific action occurs, like adding a new file to Google Drive.

Airbyte’s API gives you access to users and permissions within your workspaces, making it easy to govern their behavior and level of access. You could write a script using the Airbyte API to audit the permissions of each Airbyte user, ensuring tight data governance with your data sources.

With the Airbyte API, you can do almost anything you can do through the UI. The objects available to access (and their endpoints) include:

Applications (get, post, delete)
Sources (get, post, patch, put, delete)
Destinations (get, post, patch, put, delete)
Connections (get, post, patch, delete)
Jobs (get, post, delete)
Streams (get)
Workspaces (get, post, patch, put, delete)
Permissions (get, post, patch, put, delete)
Users (get)
Organizations (get)
Health (get)

To begin using the Airbyte API you first need to create an application on the UI, which will then allow you to generate an access token to authenticate your account with the API.

Here’s how working with the Airbyte API may look like:

Create a source: With the access token, you can create a new source. For example, to create a PostgreSQL source:

headers = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json"
}
payload = {
    "workspaceId": "YOUR_WORKSPACE_ID",
    "name": "Postgres Source",
    "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750",  # Postgres Source Definition ID
    "connectionConfiguration": {
        "host": "your_postgres_host",
        "port": 5432,
        "database": "your_database",
        "username": "your_username",
        "password": "your_password",
        "ssl": True
    }
}
response = requests.post("https://api.airbyte.com/api/v1/sources", headers=headers, json=payload)
source_id = response.json().get('sourceId')

‍Create a destination: Similarly, you can create a destination. For instance, to set up a BigQuery destination:

payload = {
    "workspaceId": "YOUR_WORKSPACE_ID",
    "name": "BigQuery Destination",
    "destinationDefinitionId": "2fdbdfea-6b31-4a9d-8d0a-5a17b5c9b1e0",  # BigQuery Destination Definition ID
    "connectionConfiguration": {
        "project_id": "your_project_id",
        "dataset_id": "your_dataset_id",
        "credentials_json": "your_credentials_json"
    }
}
response = requests.post("https://api.airbyte.com/api/v1/destinations", headers=headers, json=payload)
destination_id = response.json().get('destinationId')

Create a connection: After setting up the source and destination, establish a connection between them:

payload = {
    "sourceId": source_id,
    "destinationId": destination_id,
    "syncCatalog": {
        "streams": [
            {
                "stream": {
                    "name": "your_table_name",
                    "namespace": "public",
                    "jsonSchema": {}  # Define your schema here
                },
                "config": {
                    "syncMode": "full_refresh",
                    "destinationSyncMode": "overwrite"
                }
            }
        ]
    },
    "schedule": {
        "units": 1,
        "timeUnit": "hours"
    },
    "status": "active"
}
response = requests.post("https://api.airbyte.com/api/v1/connections/create", headers=headers, json=payload)
connection_id = response.json().get('connectionId')

For more detailed information, refer to the Airbyte API Documentation.

2. Airbyte Terraform provider

Airbyte’s Terraform provider allows you to connect to the Airbyte API and programmatically manage your syncs. This is ideal for users using Terraform to manage their data stack, as they can now manage Airbyte, collaborate with teammates, and version control their code through Terraform.

If you aren’t familiar with Terraform, it allows you to set up your infrastructure through config files, rather than doing everything manually.

Because the provider is built off the Airbyte API, all the features detailed above are available through this configuration. In addition, the Terraform provider is available for open-source, cloud, and enterprise users.

Airbyte even uses the Terraform provider internally to help manage different workspaces and connections quickly, saving time clicking around the UI.

As Alex Gronemeyer, Lead Data Engineer at Airbyte says, “We track these changes in source control and can roll back to previous configurations if needed. This gives us the peace of mind needed in a production system”.

Working with the Airbyte Terraform provider may look something like this:

Create a resource: After installing the provider, you can define sources, destinations, and connections. For example, to create a PostgreSQL source:

resource "airbyte_source" "postgres_source" {
  name       = "Postgres Source"
  source_definition_id = "decd338e-5647-4c0b-adf4-da0e75f5a750"
  workspace_id = "YOUR_WORKSPACE_ID"
  configuration = jsonencode({
    host     = "your_postgres_host"
    port     = 5432
    database = "your_database"
    username = "your_username"
    password = "your_password"
    ssl      = true
  })
}

For more details, visit the Airbyte Terraform provider documentation.

3. PyAirbyte

A more recent addition to the Airbyte’s platform set of tools is PyAirbyte.

PyAirbyte is a Python library that allows you to use Airbyte connectors directly within your Python code. With this library, you can access the most popular data sources within Airbyte and run any existing syncs.

PyAirbyte has three main components:

Source: This includes a configuration object that allows you to authenticate and connect to your Airbyte environment to read data from your connectors.
Cache: This allows you to temporarily store records from your Airbyte sources in a SQL database, such as DuckDB, Postgres, or even a cloud data warehouse like BigQuery.
Result: The result holds the data read from an Airbyte stream using the cache object. It can be accessed as a list, Pandas dataframe, or via a SQLAlchemy query, giving you options on how to use the data synced by Airbyte!

With PyAirbyte, you can programmatically run your Airbyte sync jobs within your data pipelines written in Python-based tools like Dagster or Airflow. Doing so will allow you to follow data engineering best practices, using version control and CI/CD. You can also test connections locally before deploying them to Airbyte Cloud, allowing for scalability and confidence.

The library connects to structured and unstructured data sources such as Azure Blob Storage, Github, and Notion, allowing for easier development of AI and LLM-powered applications. This way, you can easily feed data to train your models through your Python-enabled pipelines.

Here’s what the workflow looks with PyAirbyte:

Installation: Begin by installing PyAirbyte using pip:

pip install airbyte

‍Setting up a source connector: To configure a source, such as PostgreSQL, you can do it as follows:

import airbyte as ab

# Initialize the PostgreSQL source connector
source = ab.get_source(
    "source-postgres",
    install_if_missing=True,
    config={
        "host": "your_postgres_host",
        "port": 5432,
        "database": "your_database",
        "username": "your_username",
        "password": "your_password",
        "ssl": True
    }
)

‍Verifying the connection: It's essential to ensure that the source configuration is correct:

source.check()

Listing available streams: To view the data streams available from the source:

streams = source.get_available_streams()

‍Selecting and reading streams: You can select specific streams and read data into a cache:

# Select all available streams
source.select_all_streams()

# Initialize the default cache (DuckDB)
cache = ab.get_default_cache()

# Read data from the source into the cache
result = source.read(cache=cache)

‍Accessing data: Once the data is in the cache, you can load it into a Pandas DataFrame for analysis:

df = cache["your_stream_name"].to_pandas()
print(df.head())

For more detailed information and advanced configurations, refer to the PyAirbyte Documentation.

Conclusion

Programmatically managing modern data stack tools like Airbyte allows you to develop data pipelines and data products with data engineering best practices at the top of your mind. You no longer have to worry about switching between multiple UIs and making an error that can’t be reversed due to the lack of version control.

Airbyte easily integrates your favorite data ingestion tool with other technologies you already use like Python and Terraform. With the help of Airbyte API, the Airbyte Terraform provider, and PyAirbyte, you can programmatically manage your data syncs from your central code repository!

Limitless data movement with free Alpha and Beta connectors

Introducing: our Free Connector Program ->

The data movement infrastructure for the modern data teams.

Try a 14-day free trial

About the Author

Madison Schott is an Analytics Engineer, author of The ABCs of Analytics Engineering book. Madison blogs about the modern data stack on Medium and writes a newsletter on Substack.