使用dataclasses-avroschema序列化时AvroTypeException报错如何解决

问题根因 #

• 官方avro库的DatumWriter默认仅支持原生Python字典结构作为输入数据，你直接传入AvroModel子类的dataclass实例，库无法识别该数据结构，因此抛出类型不匹配错误。
• Message类的fakeMe方法存在笔误：self.bool = fake.pybool()应为self.status = fake.pybool()，虽本次初始化时你主动传入了status=True未触发该问题，但仍需修正避免后续异常。
• 你的环境同时安装了avro和avro-python3两个冲突的Avro核心库，可能引发潜在兼容问题。

解决方案 #

方案1：直接使用dataclasses-avroschema自带的序列化能力（推荐）

该库本身已经封装了完整的序列化/反序列化方法，无需自行调用底层avro库的API，代码更简洁：

from dto.temp_schema import Message

obj = Message(None, True)
obj.fakeMe()

# 直接序列化得到avro二进制数据
avro_bytes = obj.serialize()
print("serialized avro: ", avro_bytes)

# 反序列化示例
deserialized_obj = Message.deserialize(avro_bytes)
print(deserialized_obj)

方案2：适配原代码逻辑，兼容官方avro库的API

如果需要保留原有的DataFileWriter写法，按以下步骤调整即可：

1. 先卸载冲突依赖：

pip uninstall avro-python3 -y

2. 修正Message类的fakeMe方法：

def fakeMe(self):
    self.head = Head.fakeMe()
    self.status = fake.pybool() # 修正字段名

3. 调整序列化代码，传入字典结构：

import json, io as mainio
from dto.temp_schema import Message
from avro import schema, datafile, io as avroio

obj = Message(None, True)
obj.fakeMe()

schema_obj = schema.parse(json.dumps(Message.avro_schema_to_python()))

buf = mainio.BytesIO()
writer = datafile.DataFileWriter(buf, avroio.DatumWriter(), schema_obj)
writer.append(obj.asdict()) # 转成字典再传入
writer.flush()
buf.seek(0)
data = buf.read()

print("serialized avro: ", data)

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