dbt Cloud API Version Router

Build a Python utility that demonstrates proper handling of dbt Cloud's multi-version API architecture by routing requests to the correct API version.

Background

dbt Cloud uses different API versions for different resource types. Projects and environments use the V3 API, while jobs and runs use the V2 API. This design ensures backward compatibility while enabling new features.

Requirements

Create a Python module that routes API requests to the appropriate version based on resource type.

Functionality

Your utility should provide functions to:

1. Get a project by ID (V3)
2. List all environments in a project (V3)
3. Get a job by ID (V2)
4. Get a job run by ID and retrieve its status (V2)

API Structure

The dbt Cloud API uses version-specific base paths:

• V2 endpoints: /api/v2/accounts/{account_id}/...
• V3 endpoints: /api/v3/accounts/{account_id}/...

Base URL: https://{tenant}.getdbt.com

Authentication: Bearer token in Authorization header

Expected Behavior

• Automatically use the correct API version for each resource type
• Parse and return JSON responses
• Include authentication in all requests

Test Cases

• Getting a project constructs the correct V3 endpoint URL @test
• Listing environments constructs the correct V3 endpoint URL @test
• Getting a job constructs the correct V2 endpoint URL @test
• Getting a job run constructs the correct V2 endpoint URL @test

Implementation

@generates

API

class DbtCloudAPIRouter:
    """Routes dbt Cloud API requests to the appropriate API version."""

    def __init__(self, account_id: int, api_token: str, tenant: str = "cloud"):
        """Initialize the router.

        Args:
            account_id: dbt Cloud account ID
            api_token: API authentication token
            tenant: dbt Cloud tenant (default: "cloud")
        """
        pass

    def get_project(self, project_id: int) -> dict:
        """Retrieve project details using V3 API.

        Args:
            project_id: The project ID

        Returns:
            Project data
        """
        pass

    def list_environments(self, project_id: int) -> list:
        """List environments in a project using V3 API.

        Args:
            project_id: The project ID

        Returns:
            List of environments
        """
        pass

    def get_job(self, job_id: int) -> dict:
        """Retrieve job details using V2 API.

        Args:
            job_id: The job ID

        Returns:
            Job data
        """
        pass

    def get_job_run(self, run_id: int) -> dict:
        """Retrieve job run details using V2 API.

        Args:
            run_id: The run ID

        Returns:
            Job run data including status
        """
        pass

Dependencies { .dependencies }

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

Provides integration with dbt Cloud API for orchestrating data transformation workflows.

requests { .dependency }

HTTP library for making API requests.