How trigger dbt cloud job on free version

I wan't to trigger dbt cloud job with API request from a python script, however the API is not accessible for unpaid version as the error suggest : {"status": {"code": 401, "is_success": false, "user_message": "The API is not accessible to unpaid accounts.", "developer_message": null}, "data": null}.

I was trying with this code :

import requests
import os 

ACCOUNT_ID = os.environ['ACCOUNT_ID']
JOB_ID = os.environ['JOB_ID']
API_KEY = os.environ['DBT_CLOUD_API_TOKEN']

url = "https://cloud.getdbt.com/api/v2/accounts/"+ACCOUNT_ID+"/jobs/"+JOB_ID+"/run/"
headers = {
  'Authorization': 'Bearer '+API_KEY,
  'Content-Type': 'application/json'
}
response = requests.request("POST", url, headers=headers)
print(response.text)

Is there an alternative way to achieve it with the free tier version ?

Solution

You cannot use the dbt Cloud API if you are not paying the dbt Cloud service. As you can see in the pricing plan, API access is only available on the paid version.

So, one way to run dbt from a Python script would be the following:

import subprocess
from tqdm.auto import tqdm
import time

def run_dbt_command(command, project_directory):
    try:
        with tqdm(total=100, desc=f"Running dbt {command[-1]}", ncols=100) as pbar:
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, cwd=project_directory)
            
            while True:
                output = process.stdout.readline()
                if output == '' and process.poll() is not None:
                    break
                if output:
                    pbar.update(10)
                    time.sleep(0.1)  # Adjust this value to control the speed of the progress bar
                    print(output.strip())

            return_code = process.poll()
            if return_code == 0:
                print(f"dbt {command[-1]} succeeded")
            else:
                print(f"dbt {command[-1]} failed")

    except subprocess.CalledProcessError as e:
        print(f"dbt {command[-1]} failed:\n{e.stdout}")

dbt_project_path = "/path/to/your/dbt/project"
dbt_executable_path = "/path/to/your/dbt/executable"

# Run dbt commands
# run_dbt_command([dbt_executable_path, "run"], dbt_project_path)
# run_dbt_command([dbt_executable_path, "tes"], dbt_project_path)
run_dbt_command([dbt_executable_path, "build"], dbt_project_path)

Here's a brief explanation:

tqdm: this library is used to create a progress bar for loops and other iterable objects. It provides a simple, customisable way to display progress bars in various environments.
tqdm.auto: the tqdm.auto submodule is imported to automatically choose the appropriate progress bar implementation based on the environment.
Context manager: the with statement is used to create a context manager for the progress bar, which ensures that the progress bar is properly closed after the dbt command is executed.
subprocess.Popen: the subprocess.Popen method is used to run the dbt command, enabling real-time monitoring of the command's output. This allows us to update the progress bar as the command runs.
Reading output and updating progress: a while loop is used to read the output of the dbt command line by line. For each line of output, the progress bar is updated using pbar.update(10). The time.sleep(0.1) line introduces a brief pause to control the speed of the progress bar. You can adjust this value to make the progress bar faster or slower.
Checking return code: After the dbt command is completed, the return_code is checked to determine if the command succeeded or failed. This is done using process.poll(), which returns the return code of the command. If the return code is 0, the command succeeded, otherwise it failed.

So, long story short, you are just running the dbt commands that you specify in the script, and outputting the dbt logs in the CLI.