from dbt.cli.main import dbtRunner, dbtRunnerResult
# initialize
dbt = dbtRunner()
# create CLI args as a list of strings
cli_args = ["run", "--select", "tag:my_tag"]
# run the command
res: dbtRunnerResult = dbt.invoke(cli_args)
# inspect the results
for r in res.result:
print(f"{r.node.name}: {r.status}")
1条答案
按热度按时间8i9zcol21#
更新:v1。5来了!
通过dbt的v1.5,我们得到了一个稳定的、官方支持的Python API,用于调用dbt操作;此API与CLI具有功能对等性。
关于docs:
关于
dbt.invoke
返回的工件的稳定性有一些警告;阅读文档了解更多细节。原始应答
(As 2023年1月)目前还没有一个用于dbt的公共Python API。预计在v1中。5,应该在几个月内。
现在,您最安全的选择是使用CLI。如果您不想使用
subprocess
,CLI现在使用Click,而Click提供了一个运行程序,您可以使用它来调用Click命令。它通常用于测试,但我认为它也适用于您的用例。CLI命令在这里。它看起来像这样:您也可以像在测试套件中那样调用dbt,使用
run_dbt
。