如何在Python脚本或程序中运行DBT模型？

方法一：Python调用DBT CLI（推荐，稳定易维护） #

这是最常用的方案，无需额外依赖特殊库，只要你已经安装了DBT即可。利用Python的subprocess模块直接执行DBT命令：

import subprocess
import os

# 切换到DBT项目根目录（如果脚本不在项目根目录）
os.chdir("/path/to/your/dbt/project")

# 执行dbt run命令，可根据需求调整参数
run_result = subprocess.run(
    ["dbt", "run"],
    capture_output=True,
    text=True
)

# 打印运行结果
print("标准输出:\n", run_result.stdout)
print("错误输出:\n", run_result.stderr)
print("返回码:", run_result.returncode)

常用参数示例：

• 运行指定模型：["dbt", "run", "--models", "your_target_model"]
• 指定生产环境：["dbt", "run", "--target", "production"]
• 自定义配置文件路径：["dbt", "run", "--profiles-dir", "/path/to/your/profiles"]

方法二：使用DBT Python API（库集成方式） #

DBT Core本身提供了Python API，可以直接在脚本中调用，无需通过CLI。这种方式能直接获取结构化的运行结果，适合深度集成场景：

第一步：安装依赖 #

pip install dbt-core dbt-postgres  # 根据你的数据库替换适配器，比如dbt-bigquery

第二步：脚本示例 #

from dbt.cli.main import dbtRunner, dbtRunnerResult

# 初始化DBT运行器
runner = dbtRunner()

# 定义DBT命令参数（与CLI参数完全一致）
cli_args = ["run", "--models", "your_target_model"]

# 执行命令
result: dbtRunnerResult = runner.invoke(cli_args)

# 处理结果
if result.success:
    print("DBT运行成功")
    for artifact in result.result:
        print(f"模型 {artifact.node.name} 状态: {artifact.status}")
else:
    print("DBT运行失败")
    print("错误详情:", result.exception)

注意事项 #

• 确保脚本运行环境能访问到DBT的profiles.yml，可通过设置环境变量DBT_PROFILES_DIR指定配置文件路径
• API参数与CLI完全兼容，可参考DBT官方文档调整参数

内容的提问来源于stack exchange，提问作者Brian Postow