如何实现PyDantic模型到gRPC通信的直接转换？

需求背景 #

我们计划落地的架构核心逻辑：

• 所有服务的数据模型采用PyDantic定义，复用其内置验证机制
• 基于这些模型生成gRPC的pb2文件，作为服务间内部通信的标准
• 用FastAPI搭建api-gateway，对外暴露接口（通过FastAPI的OpenAPI装饰器自动生成文档），内部逻辑直接调用对应gRPC服务

该架构的核心优势：

1. 以PyDantic模型作为唯一数据源，实现统一验证与建模
2. 自动生成服务间gRPC通信能力
3. 自动生成对外OpenAPI文档，精准管控暴露的接口内容

当前实现流程存在冗余：微服务层通过FastAPI生成JSON Schema → 用openapi-generator（Java）转换为Proto文件 → 借助grpc_tools.protoc生成Python客户端。问题在于PyDantic本身已是结构化定义，FastAPI仅被用来生成OpenAPI而非提供HTTP服务，多步转换过于繁琐，因此需要直接从PyDantic转换为gRPC通信的方案。

可行解决方案 #

方法1：使用pydantic-protobuf直接转换模型 #

这是最直接的方案，该库可以将PyDantic模型一键转换为Protobuf定义，无需中间步骤：

1. 安装依赖：

   pip install pydantic-protobuf
   

2. 在PyDantic模型上添加@protobuf装饰器：

   from pydantic import BaseModel
   from pydantic_protobuf import protobuf
   
   @protobuf
   class User(BaseModel):
       user_id: int
       username: str
       email: str | None = None
   

3. 生成.proto文件：

   pydantic-protobuf generate --module your_module.models --output ./proto
   

4. 用grpc_tools.protoc基于生成的proto文件生成gRPC客户端/服务端代码：

   python -m grpc_tools.protoc -I./proto --python_out=./grpc_code --grpc_python_out=./grpc_code ./proto/user.proto
   

方法2：自定义代码生成脚本（灵活控制） #

如果需要自定义gRPC服务方法、包名或字段映射规则，可以编写简单的Python脚本直接生成.proto文件：

from pydantic import BaseModel
from typing import get_args, get_origin

# 映射PyDantic类型到Protobuf类型
TYPE_MAP = {
    int: "int32",
    str: "string",
    bool: "bool",
    float: "float"
}

def pydantic_to_proto(model: type[BaseModel], service_name: str) -> str:
    proto_content = []
    proto_content.extend([
        'syntax = "proto3";',
        f'package {service_name};',
        ''
    ])

    # 生成消息定义
    proto_content.append(f'message {model.__name__} {{')
    for field_idx, (field_name, field) in enumerate(model.model_fields.items(), start=1):
        field_type = field.annotation
        # 处理Optional类型
        is_optional = False
        if get_origin(field_type) is Union:
            args = get_args(field_type)
            if len(args) == 2 and args[1] is type(None):
                field_type = args[0]
                is_optional = True
        # 映射类型
        proto_type = TYPE_MAP.get(field_type, "string")
        if is_optional:
            proto_content.append(f'  optional {proto_type} {field_name} = {field_idx};')
        else:
            proto_content.append(f'  {proto_type} {field_name} = {field_idx};')
    proto_content.append('}')

    # 生成服务定义（可根据业务需求自定义）
    proto_content.extend([
        '',
        f'service {service_name.capitalize()}Service {{',
        f'  rpc Get{model.__name__}(Get{model.__name__}Request) returns ({model.__name__});',
        f'  rpc Create{model.__name__}({model.__name__}) returns ({model.__name__});',
        '}'
    ])

    # 生成请求消息
    proto_content.extend([
        '',
        f'message Get{model.__name__}Request {{',
        f'  int32 id = 1;',
        '}'
    ])

    return '\n'.join(proto_content)

# 使用示例
class User(BaseModel):
    user_id: int
    username: str
    email: str | None = None

# 生成proto内容并写入文件
with open('./proto/user.proto', 'w') as f:
    f.write(pydantic_to_proto(User, "user_service"))

方法3：结合grpclib与PyDantic直接序列化 #

如果不想生成pb文件，可借助grpclib在gRPC服务中直接使用PyDantic模型进行序列化/反序列化，业务逻辑无需重复定义模型：

1. 安装依赖：

   pip install grpclib pydantic
   

2. 服务端与客户端实现示例：

   from grpclib.server import Server, Stream
   from grpclib.client import Channel
   from pydantic import BaseModel
   import asyncio
   
   # PyDantic业务模型
   class User(BaseModel):
       user_id: int
       username: str
       email: str | None = None
   
   # 提前用protoc生成的pb2和grpc代码（proto定义仅需基础消息结构）
   from user_pb2 import UserRequest, UserResponse
   from user_grpc import UserServiceBase, UserServiceStub
   
   # gRPC服务端实现
   class UserService(UserServiceBase):
       async def GetUser(self, stream: Stream[UserRequest, UserResponse]) -> None:
           request = await stream.recv_message()
           # 用PyDantic处理业务逻辑（验证、数据转换等）
           user_data = {"user_id": request.user_id, "username": "test_user", "email": "test@example.com"}
           user = User(**user_data)
           # 转换为proto响应
           response = UserResponse(**user.model_dump())
           await stream.send_message(response)
   
   # gRPC客户端调用
   async def client_call():
       async with Channel("localhost", 50051) as channel:
           stub = UserServiceStub(channel)
           request = UserRequest(user_id=1)
           response = await stub.GetUser(request)
           # 将proto响应转换为PyDantic模型
           user = User(**response.__dict__)
           print(user)
   
   if __name__ == "__main__":
       # 启动服务端
       server = Server([UserService()])
       loop = asyncio.get_event_loop()
       loop.create_task(server.start("localhost", 50051))
       # 执行客户端调用
       loop.run_until_complete(client_call())
   

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