dbt_core_programmatic_command_invocation_with_dbtrunner.py

1from dbt.cli.main import dbtRunner, dbtRunnerResult
2
3# Initialize the dbtRunner
4dbt = dbtRunner()
5
6# Create a list of arguments for the dbt command
7# This example runs 'dbt run' with a specific target
8cli_args = ["run", "--target", "dev"]
9
10# Execute the command
11res: dbtRunnerResult = dbt.invoke(cli_args)
12
13# Inspect the results
14if res.success:
15    for r in res.result:
16        print(f"{r.node.name}: {r.status}")
17else:
18    if res.exception:
19        print(f"Command failed with exception: {res.exception}")
20    else:
21        print("Command failed, check dbt logs for details")