dbt宏调用Python subprocess模块遇'imprt未定义'编译错误求助

错误原因 #

你的问题出在dbt宏的运行环境和你单独编写的jinja2渲染脚本完全隔离。你在脚本中传递的imprt参数，dbt执行宏时根本无法访问，因此宏里的imprt被判定为未定义变量。

正确实现方式 #

dbt从1.3版本开始支持在宏中直接使用{% python %}块编写Python代码，不需要额外的渲染脚本。直接在宏内导入subprocess并执行命令即可：

修改后的宏代码 #

{% macro check_select_arg(run_all=False, dbt_command='dbt run') %}
{% do log("ON START ✅", info=True) %}

{% if run_all %}
    {% python %}
    import subprocess
    # 执行dbt命令并捕获输出
    result = subprocess.run(
        dbt_command, 
        shell=True, 
        stdout=subprocess.PIPE, 
        stderr=subprocess.PIPE, 
        text=True
    )
    # 将命令输出打印到dbt日志
    print(f"Command execution stdout:\n{result.stdout}")
    if result.stderr:
        print(f"Command execution stderr:\n{result.stderr}")
    {% endpython %}

{% elif not invocation_args_dict.select
        and target.name not in ['prod']
        and invocation_args_dict.which in ['build', 'run', 'test', 'source', 'snapshot', 'seed'] %}
    {{ exceptions.raise_compiler_error("❌ Error: You must provide at least one select argument") }}

{% endif %}

{% endmacro %}

运行说明 #

1. 确保你的dbt版本≥1.3（支持Python块功能）
2. 删除你之前写的单独Python渲染脚本，它对dbt宏的执行没有作用
3. 继续使用原命令运行：

dbt run-operation check_select_arg --arg "{run_all: True, dbt_command: 'dbt run'}"

注意事项 #

• 执行subprocess时，dbt运行的环境需能正常找到dbt命令（可使用绝对路径避免环境变量问题）
• 若需要处理命令执行结果，可在{% python %}块内添加逻辑判断（比如根据returncode判断是否执行成功）

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