Python中无预定义.proto文件，JSON转Protobuf序列化消息及动态实现咨询

一、无需.proto文件直接将JSON转为Protobuf序列化消息 #

可以通过Protobuf的**动态消息（Dynamic Message）**能力实现，完全不需要提前编写.proto文件或预编译类。Python的google.protobuf库提供了descriptor_pool和message_factory模块，支持在运行时根据动态Schema构建消息描述符，进而生成可序列化的消息对象。

核心思路是先将传入的Schema（比如BigQuery的Schema结构）映射为Protobuf的字段描述符，再构建消息描述符，最后生成动态消息类来处理JSON数据。示例代码如下：

from google.protobuf import descriptor, descriptor_pool, message_factory, json_format

# 模拟从外部获取的动态Schema（比如BigQuery返回的Schema）
dynamic_schema = [
    {"name": "id", "type": "INTEGER", "mode": "REQUIRED"},
    {"name": "username", "type": "STRING", "mode": "NULLABLE"},
    {"name": "is_valid", "type": "BOOLEAN", "mode": "NULLABLE"}
]

# 构建Protobuf字段描述符列表
field_descriptors = []
bq_to_proto_type = {
    "INTEGER": descriptor.FieldDescriptor.TYPE_INT64,
    "STRING": descriptor.FieldDescriptor.TYPE_STRING,
    "BOOLEAN": descriptor.FieldDescriptor.TYPE_BOOL
}

for idx, field in enumerate(dynamic_schema):
    # 确定字段是否必填
    field_label = descriptor.FieldDescriptor.LABEL_REQUIRED if field["mode"] == "REQUIRED" else descriptor.FieldDescriptor.LABEL_OPTIONAL
    field_desc = descriptor.FieldDescriptor(
        name=field["name"],
        full_name=f"DynamicData.{field['name']}",
        index=idx,
        number=idx + 1,  # Protobuf字段编号从1开始
        type=bq_to_proto_type[field["type"]],
        label=field_label,
        containing_type=None
    )
    field_descriptors.append(field_desc)

# 创建消息描述符
pool = descriptor_pool.DescriptorPool()
message_desc = pool.AddDescriptor(descriptor.Descriptor(
    name="DynamicData",
    full_name="DynamicData",
    fields=field_descriptors,
    syntax="proto3"
))

# 生成动态消息类
DynamicDataMsg = message_factory.GetMessageClass(message_desc)

# 将JSON转为Protobuf消息并序列化
json_input = '{"id": 789, "username": "demo_user", "is_valid": true}'
msg_instance = DynamicDataMsg()
json_format.Parse(json_input, msg_instance)
serialized_bytes = msg_instance.SerializeToString()

# 后续可将serialized_bytes用于BigQuery的Protobuf格式插入

这种方案轻量高效，无需依赖外部编译工具，非常适合未知或频繁变化的Schema场景。

二、运行时动态生成.proto文件并编译使用Python类 #

这个方案也可行，但流程更复杂，需要在运行时完成.proto文件生成、编译、动态导入三个步骤，适合必须使用预编译类的场景（比如某些框架强制要求）。

核心步骤：

1. 根据动态Schema生成符合proto3语法的文本内容
2. 将文本保存为临时.proto文件
3. 调用protoc编译器编译生成Python模块
4. 动态导入模块并使用消息类

示例代码：

import os
import subprocess
import importlib.util
from google.protobuf import json_format

# 模拟动态Schema
dynamic_schema = [
    {"name": "order_id", "type": "INTEGER", "mode": "REQUIRED"},
    {"name": "product_name", "type": "STRING", "mode": "NULLABLE"},
    {"name": "amount", "type": "FLOAT", "mode": "NULLABLE"}
]

# 生成proto3格式的文件内容
proto_text = """syntax = "proto3";

message DynamicOrder {
"""
bq_to_proto_type = {
    "INTEGER": "int64",
    "STRING": "string",
    "FLOAT": "double",
    "BOOLEAN": "bool"
}

for idx, field in enumerate(dynamic_schema):
    required_flag = "required " if field["mode"] == "REQUIRED" else ""
    proto_text += f"  {required_flag}{bq_to_proto_type[field['type']]} {field['name']} = {idx + 1};\n"
proto_text += "}\n"

# 保存临时.proto文件
temp_proto_path = "/tmp/dynamic_order.proto"
with open(temp_proto_path, "w") as f:
    f.write(proto_text)

# 编译.proto生成Python代码
output_dir = "/tmp"
subprocess.run(
    ["protoc", f"--python_out={output_dir}", temp_proto_path],
    check=True,
    capture_output=True
)

# 动态导入生成的模块
module_path = os.path.join(output_dir, "dynamic_order_pb2.py")
spec = importlib.util.spec_from_file_location("dynamic_order_pb2", module_path)
order_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(order_module)

# 使用生成的类处理JSON数据
json_input = '{"order_id": 1001, "product_name": "Laptop", "amount": 999.99}'
order_msg = order_module.DynamicOrder()
json_format.Parse(json_input, order_msg)
serialized_bytes = order_msg.SerializeToString()

# 清理临时文件（可选）
os.remove(temp_proto_path)
os.remove(module_path)

注意事项：

• 运行环境必须安装protoc编译器，否则无法完成编译
• 临时文件需做好管理，避免多线程场景下的文件冲突
• 编译过程有额外性能开销，不适合Schema高频变化的场景

方案选型建议 #

• 优先选择动态消息方案：无需依赖外部工具，性能更高，适配灵活，完全满足未知Schema的需求
• 动态编译方案仅在必须使用预编译类的场景下考虑，比如部分第三方库强制要求传入编译后的Protobuf类对象

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