如何创建字段名不同且包含外键关联字段的Pydantic模型？

注意：原SQLAlchemy模型中Session类的__tablename__写错为run_dir，会导致外键关联失败，需修正为session，正确模型如下：

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class RunDir(Base):
    __tablename__ = 'run_dir'

    id = Column(Integer, primary_key=True)
    path = Column(String(1024))
    session_id = Column(Integer, ForeignKey('session.id'))
    session = relationship('Session', backref='run_dirs') 

class Session(Base):
    __tablename__ = 'session'  # 修正表名以匹配外键关联

    id = Column(Integer, primary_key=True)
    name = Column(String(1024))
    last_time_run = Column(String(1024))

一、查询用响应模型（包含关联Session字段） #

要实现自定义字段名并提取关联模型的字段，可借助Pydantic的Field(alias)做字段映射，结合validator从关联对象中提取数据，同时开启orm_mode支持从SQLAlchemy实例直接转换：

from pydantic import BaseModel, Field, validator

class RunDirResponse(BaseModel):
    # 用alias映射SQLAlchemy原字段名
    run_dir_id: int = Field(alias='id')
    path_to_dir: str = Field(alias='path')
    # 关联Session的自定义字段
    session_name: str
    run_session_last_time: str  # 原SQLAlchemy中last_time_run为String类型，若需转为int可添加类型转换逻辑

    class Config:
        orm_mode = True  # 开启ORM模式，支持从SQLAlchemy实例生成模型

    # 从关联的session对象提取名称
    @validator('session_name', pre=True)
    def extract_session_name(cls, _, values):
        if session_obj := values.get('session'):
            return session_obj.name
        raise ValueError("关联的Session对象不存在")

    # 从关联的session对象提取最后运行时间
    @validator('run_session_last_time', pre=True)
    def extract_last_run_time(cls, _, values):
        if session_obj := values.get('session'):
            # 如需转为int，取消下方注释并处理异常
            # try:
            #     return int(session_obj.last_time_run)
            # except (ValueError, TypeError):
            #     raise ValueError("last_time_run无法转换为整数")
            return session_obj.last_time_run
        raise ValueError("关联的Session对象不存在")

使用方式： #

查询时需提前加载关联的Session对象（避免懒加载导致的数据库查询异常），示例：

from sqlalchemy.orm import joinedload

# 加载关联的Session
run_dir_instance = db.query(RunDir).options(joinedload(RunDir.session)).first()
# 转换为Pydantic模型
response_model = RunDirResponse.from_orm(run_dir_instance)

二、创建/更新用请求模型 #

若用于接收前端请求数据并生成SQLAlchemy实例，可直接映射字段并提供转换方法：

class CreateRunDirRequest(BaseModel):
    path_to_dir: str
    session_id: int  # 直接接收外键ID用于关联

    def to_run_dir_orm(self):
        return RunDir(
            path=self.path_to_dir,
            session_id=self.session_id
        )

使用方式： #

# 假设请求数据已解析为CreateRunDirRequest实例
request_model = CreateRunDirRequest(path_to_dir="/tmp/run1", session_id=1)
# 转换为SQLAlchemy实例并保存
db.add(request_model.to_run_dir_orm())
db.commit()

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