dbt_core_programmatic_runner_invoke_with_result_inspection.py

1from dbt.cli.main import dbtRunner, dbtRunnerResult
2
3# Initialize the dbt runner
4dbt = dbtRunner()
5
6# Create a list of arguments (similar to what you would type in the CLI)
7# Example: 'dbt run --select my_model'
8cli_args = ["run", "--select", "my_first_dbt_model"]
9
10# Execute the command
11res: dbtRunnerResult = dbt.invoke(cli_args)
12
13# Inspect the results
14if res.success:
15    for result in res.result:
16        print(f"Status for {result.node.name}: {result.status}")
17else:
18    if res.exception:
19        print(f"An error occurred: {res.exception}")
20    else:
21        print("The dbt command failed.")