Python脚本中从dbt获取查询结果的优雅方案问询

问题核心 #

使用dbt 1.8的Python API执行查询时，dbtRunnerResult无法直接返回查询结果，当前依赖解析日志的临时方案不够优雅，且希望复用dbt的数据库连接，避免单独维护数据库客户端。

解决方案 #

1. 修改dbt宏，让其返回查询结果 #

将原宏改为执行查询后返回结构化结果，而不是仅打印日志：

{% macro execute_sql(query) %}
{% set results = run_query(query) %}
{% set result_list = [] %}
{% for row in results %}
    {% do result_list.append(row.as_dict()) %}  <!-- 返回字典格式，更易处理多列结果 -->
{% endfor %}
{{ return(result_list) }}
{% endmacro %}

如果仅需要单列值，也可以简化为：

{% macro execute_sql(query) %}
{% set results = run_query(query) %}
{{ return(results.rows|map(attribute='values')|list) }}
{% endmacro %}

2. Python脚本直接获取返回结果 #

通过dbtRunner.invoke()调用run-operation后，dbtRunnerResult.result会直接拿到宏返回的结果：

from dbt.cli.main import dbtRunner, dbtRunnerResult
import json

# 初始化dbt runner
dbt = dbtRunner()

# 定义要执行的SQL查询
sql_query = "SELECT COUNT(*) FROM dbt.model_x"

# 调用自定义宏执行查询
run_result: dbtRunnerResult = dbt.invoke([
    'run-operation',
    'execute_sql',
    '--args',
    json.dumps({"query": sql_query})
])

if run_result.success:
    print("查询执行成功")
    # 直接获取结构化结果
    query_results = run_result.result
    for row in query_results:
        print(row)  # 字典格式示例: {'count': 123}
else:
    print("查询执行失败")
    print(run_result.exception)

关键原理 #

dbt 1.8的run-operation命令会将宏中return的值作为执行结果返回给Python API的dbtRunnerResult.result属性，无需再解析日志输出。这种方式完全复用dbt的数据库连接配置，无需单独维护数据库客户端及认证信息。

注意事项 #

• 确保宏返回的是可序列化的数据结构（列表、字典、元组等），避免返回dbt内部对象
• 如果查询结果为空，run_result.result会返回空列表
• 若需要处理大量数据，建议分批获取或限制查询结果量，避免内存占用过高

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