如何通过dbt-core Python库获取compiled_sql或写入manifest文件

方案1：将compiled_sql写入manifest文件 #

要让compiled_sql字段出现在target/manifest.json里，核心是先触发dbt的编译流程：

• 命令行方式：直接执行dbt compile，编译完成后打开target/manifest.json，每个模型条目下的compiled_sql就是编译后的SQL代码
• Python代码触发编译：

from dbt.task.compile import CompileTask
from dbt.config import RuntimeConfig

# 替换为你的dbt项目路径
project_dir = "./your-dbt-project"
# 初始化编译配置
config = RuntimeConfig.from_args(args=["compile"], project_dir=project_dir)
# 执行编译任务
task = CompileTask(config)
task.run()

执行后target/manifest.json会自动填充所有模型的compiled_sql字段。

方案2：Python代码直接访问compiled_sql #

如果不想依赖manifest文件，可通过dbt的编译API直接获取：

from dbt.config import Project
from dbt.parser.manifest import ManifestLoader
from dbt.compilation import Compiler

# 加载dbt项目
project = Project.from_project_root("./your-dbt-project")
# 加载未编译的manifest
manifest_loader = ManifestLoader(project)
raw_manifest = manifest_loader.load()
# 编译所有模型
compiler = Compiler(project, raw_manifest)
compiled_manifest = compiler.compile()

# 示例：获取指定模型的compiled_sql
model_unique_id = f"model.{project.project_name}.your_model_name"
if model_unique_id in compiled_manifest.nodes:
    compiled_sql = compiled_manifest.nodes[model_unique_id].compiled_sql
    print(compiled_sql)

# 结合networkx图使用（假设nx_graph是你的依赖图）
for node_id in nx_graph.nodes:
    if node_id in compiled_manifest.nodes:
        sql = compiled_manifest.nodes[node_id].compiled_sql
        # 这里可以对SQL做存储、分析等操作

注意点 #

• 确保你的dbt-core版本包含你提到的功能（建议使用>=1.0.0的稳定版本）
• 运行代码时要保证工作目录正确，或显式指定project_dir路径
• 编译过程会自动处理模型依赖，无需手动排序

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