dbt Cloud Connection Validator

Build a connection validation utility for dbt Cloud that verifies API connectivity and retrieves basic account information.

Requirements

Create a Python module that provides functionality to:

1. Establish a connection to the dbt Cloud API using provided credentials
2. Test the connection to verify that the API credentials are valid and the connection is working
3. Retrieve account information to confirm successful authentication and return the list of authorized accounts

The utility should accept:

• An API token for authentication
• An optional account ID (if multiple accounts are authorized)
• An optional tenant/host configuration for custom dbt Cloud instances

Expected Behavior

Connection Establishment

• Accept API credentials (token, account ID, tenant/host)
• Initialize a connection to the dbt Cloud API
• Support both default cloud.getdbt.com and custom tenant hosts

Connection Testing

• Verify that the provided credentials are valid
• Confirm API accessibility
• Return a boolean indicating connection success

Account Retrieval

• Fetch the list of accounts the API token has access to
• Return account details including account IDs and names
• Handle cases where no accounts are accessible

Test Cases

• Given valid API credentials, test_connection should return True @test
• Given valid credentials, get_accounts should return a list with at least one account containing 'id' and 'name' fields @test
• Given invalid API credentials, test_connection should return False or raise an authentication error @test

Implementation

@generates

API

def create_connection(token: str, account_id: int = None, tenant: str = None):
    """
    Create a connection to dbt Cloud API.

    Args:
        token: API authentication token
        account_id: Optional account ID for multi-account scenarios
        tenant: Optional custom tenant/host for private dbt Cloud instances

    Returns:
        A connection object that can be used for API interactions
    """
    pass

def test_connection(connection) -> bool:
    """
    Test if the connection to dbt Cloud API is valid.

    Args:
        connection: The connection object to test

    Returns:
        True if connection is valid, False otherwise
    """
    pass

def get_accounts(connection) -> list:
    """
    Retrieve list of accounts accessible with the current credentials.

    Args:
        connection: The connection object to use

    Returns:
        List of dictionaries containing account information (id, name, etc.)
    """
    pass

Dependencies { .dependencies }

apache-airflow-providers-dbt-cloud { .dependency }

Provides dbt Cloud API integration capabilities for Apache Airflow.