dbt-server 简单说明-526互联

dbt-server 是dbt 团队对于dbt-rpc 调整之后开源的dbt api 服务，内部基于了dbt-core 提供的cli 编程调用能力

使用到的一些框架

fastapi 提供api 能力
dbt core 进行dbt 任务的执行
celery 基于celery 的任务调度（代码中的worker）

dbt core cli 调用

参考代码

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}")

状态服用
可以基于Manifest对象

 
from dbt.cli.main import dbtRunner, dbtRunnerResult

from dbt.contracts.graph.manifest import Manifest
 
# use 'parse' command to load a Manifest

res: dbtRunnerResult = dbtRunner().invoke(["parse"])

manifest: Manifest = res.result
 
# introspect manifest

# e.g. assert every public model has a description

for node in manifest.nodes.values():

    if node.resource_type == "model" and node.access == "public":

        assert node.description != "", f"{node.name} is missing a description"
 
# reuse this manifest in subsequent commands to skip parsing

dbt = dbtRunner(manifest=manifest)

cli_args = ["run", "--select", "tag:my_tag"]

res = dbt.invoke(cli_args)