如何在dbt程序化调用中获取回调函数的返回值？

在dbt-core 1.7.7里，程序化调用时的回调函数返回值不会被dbt保存，所以靠返回值拿数据是行不通的。不想用全局字典的副作用写法，可以试试下面两种更优雅的方式：

方式一：用闭包封装数据 #

闭包能让回调函数访问外部变量，不用全局变量，代码更整洁：

from dbt.cli.main import dbtRunner, dbtRunnerResult
from dbt.events.base_types import EventMsg

def create_stats_collector():
    stats_data = []
    def callback(event: EventMsg):
        if event.info.name == 'StatsLine':
            stats_data.append(event.data)
    return callback, stats_data

# 创建回调函数和存储数据的列表
stats_callback, collected_stats = create_stats_collector()

cli_args = ['run', '-m', 'my_model']
dbt = dbtRunner(callbacks=[stats_callback])
res = dbt.invoke(cli_args)

# 直接用collected_stats获取所有StatsLine数据
print(collected_stats)

方式二：用类管理状态 #

如果后续要扩展收集更多事件，用类封装会更清晰：

from dbt.cli.main import dbtRunner, dbtRunnerResult
from dbt.events.base_types import EventMsg

class StatsCollector:
    def __init__(self):
        self.stats = []
    
    def callback(self, event: EventMsg):
        if event.info.name == 'StatsLine':
            self.stats.append(event.data)

# 实例化收集器
collector = StatsCollector()

cli_args = ['run', '-m', 'my_model']
dbt = dbtRunner(callbacks=[collector.callback])
res = dbt.invoke(cli_args)

# 通过collector.stats拿到收集到的数据
print(collector.stats)

为什么回调返回值没用？ #

dbt的回调机制设计就是只执行回调函数，不会捕获或保存任何返回值，所以你在回调里返回的event.data不会被存到dbtRunnerResult或其他dbt返回的对象里，这就是你找不到返回值的原因。

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