如何将DataFrame生成的字典反序列化为faust Record实例？

当你在Faust Record中用函数（如json.loads、pd.Timestamp）作为字段类型转换器时，直接使用cls(**row)、cls.loads(row)或cls.from_data(row)都不会自动触发这些转换器——因为这些方法默认仅处理原生类型，不会执行自定义字段转换逻辑。

通用解决方案 #

核心思路是手动遍历Faust Record的字段定义，对每个字段应用对应的转换器，再实例化模型。利用Faust Record内置的_fields属性可以获取所有字段的元数据（包括你定义的转换器函数）。

通用转换函数

import faust

def dict_to_faust_model(row: dict, cls: faust.Record) -> faust.Record:
    processed_row = {}
    # 遍历模型所有字段定义
    for field_name, field in cls._fields.items():
        value = row.get(field_name, field.default)
        # 如果字段类型是可调用的转换器，执行转换
        if callable(field.type):
            try:
                processed_row[field_name] = field.type(value)
            except Exception as e:
                raise ValueError(f"转换字段{field_name}失败，值: {value}，错误: {str(e)}")
        else:
            processed_row[field_name] = value
    return cls(**processed_row)

在你的代码中使用

# cols是类型声明的字段列表，cls = MyModel
cols = get_class_variables(cls)

# 返回指定列的DataFrame
df = client.select(
    f'SELECT {", ".join(cols)} FROM ... WHERE ...', columns=cols)

# 转换为list[MyModel]类型
values = [dict_to_faust_model(row, cls) for row in df.to_dict(orient='records')]

为什么之前的方法无效 #

• cls(**row)：直接传入字典实例化仅做参数传递，跳过了Faust的字段转换流程，不会触发自定义转换器。
• cls.loads(row)：该方法用于解析序列化后的字节数据（如JSON字符串），不支持直接处理Python字典。
• cls.from_data(row)：内部逻辑等价于cls(**row)，同样不会执行自定义字段转换。

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