微服务架构下使用FastAPI如何避免Pydantic模型重复？

虽然有观点认为跨服务共享数据模型会提升服务间耦合度，但如果确实需要实现模型复用以避免重复定义，以下是针对FastAPI+Kubernetes场景的具体实现方案：

1. 构建私有共享Python包 #

这是最常规的解耦复用方式，将所有跨服务共享的Pydantic模型封装成独立的Python包，供网关和各后端服务依赖安装：

步骤： #

• 创建单独的Git仓库（如service-shared-models），编写共享模型代码：

  # service-shared-models/src/shared_models/campaign.py
  from pydantic import BaseModel
  
  class CreateCampaign(BaseModel):
      name: str
  

• 将包发布到私有PyPI仓库（如公司内部的DevPi、Artifactory），或直接在Docker构建时从Git拉取安装。
• 在网关和campaign服务的requirements.txt中添加依赖：

  service-shared-models @ git+https://your-git-repo/service-shared-models.git@v1.0.0
  

• K8s镜像构建时，确保安装该依赖（Dockerfile示例）：

  FROM python:3.11-slim
  WORKDIR /app
  COPY requirements.txt .
  RUN pip install --no-cache-dir -r requirements.txt
  COPY . .
  CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
  

• 服务中直接导入使用：

  # campaign服务和网关服务中
  from shared_models.campaign import CreateCampaign
  

2. 基于OpenAPI自动生成模型 #

利用FastAPI自动生成OpenAPI规范的特性，从后端服务的OpenAPI文档自动生成网关所需的Pydantic模型：

步骤： #

• 安装模型生成工具，如datamodel-code-generator：

  pip install datamodel-code-generator
  

• 在网关服务的构建流程中（如Dockerfile或CI脚本），拉取campaign服务的OpenAPI规范并生成模型：

  # 假设campaign服务在K8s集群内的地址为http://campaign-service:8000
  datamodel-codegen --input http://campaign-service:8000/openapi.json --output app/generated_models/campaign.py
  

• 网关中导入生成的模型使用：

  from app.generated_models.campaign import CreateCampaign
  

• 注意：可在CI/CD流程中添加模型生成步骤，确保网关模型与后端服务同步更新。

3. 使用Git子模块共享模型文件 #

如果不想维护独立的Python包，可将共享模型文件作为Git子模块嵌入到网关和后端服务的仓库中：

步骤： #

• 创建单独的Git仓库存储共享模型（如shared-models），存放campaign.py等模型文件。
• 在网关和campaign服务的Git仓库中添加子模块：

  git submodule add https://your-git-repo/shared-models.git app/shared_models
  

• 构建Docker镜像时，确保拉取子模块内容（Dockerfile示例）：

  FROM python:3.11-slim
  WORKDIR /app
  RUN apt-get update && apt-get install -y git
  COPY . .
  RUN git submodule update --init --recursive
  RUN pip install --no-cache-dir -r requirements.txt
  CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
  

• 服务中直接导入子模块内的模型：

  from app.shared_models.campaign import CreateCampaign
  

4. 运行时动态生成模型（应急方案） #

如果以上方案都无法快速落地，可通过解析后端服务的OpenAPI规范，在运行时动态生成Pydantic模型（不推荐用于生产环境，无静态类型检查）：

示例代码： #

import httpx
from pydantic import create_model

async def get_campaign_models():
    async with httpx.AsyncClient() as client:
        resp = await client.get("http://campaign-service:8000/openapi.json")
        openapi_spec = resp.json()
        create_campaign_schema = openapi_spec["components"]["schemas"]["CreateCampaign"]
        
        # 动态生成Pydantic模型
        CreateCampaign = create_model(
            "CreateCampaign",
            **{field: (schema["type"], ...) for field, schema in create_campaign_schema["properties"].items()}
        )
        return CreateCampaign

# 网关端点中使用
@router.post('')
async def create_campaign():
    CreateCampaign = await get_campaign_models()
    # 后续请求校验逻辑...

内容的提问来源于Stack Exchange，提问作者Dương Tiểu Dong