如何在SQLAlchemy+Pydantic中预加载关联字段数据？

问题场景 #

定义了以下SQLAlchemy模型：

class Address(Base):
    event_id: Mapped[int] = mapped_column(
        sa.Integer,
        sa.ForeignKey('events.id'),
        nullable=False
    )
    event = relationship(
        'Event',
        back_populates='address',
        uselist=False
    )
    # 补充：需包含对应Schema的字段，否则无法映射
    country: Mapped[str]
    city: Mapped[str]
    street: Mapped[str]
    build: Mapped[str]

class Event(Base):
    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    address = relationship(
        'Address',
        uselist=False,
        back_populates='event',
    )

通过event = Event(**data)创建实例并提交会话后，调用EventSchema.from_orm(event_instance)时出现value is not a valid dict (type=type_error.dict)错误。对应的Pydantic Schema定义：

class AddressSchema(BaseModel):
    country: str
    city: str
    street: str
    build: str

class EventSchema(BaseModel):
    id: int
    address: Optional[AddressSchema] = Field(...)

期望得到的响应格式：

{"id": 1, "address": {"country": "foocountry", "street": "foostreet"}}

报错原因 #

SQLAlchemy关系默认采用懒加载（lazy='select'），会话关闭后event_instance.address是未初始化的关系代理对象，并非实际的Address实例。Pydantic无法将该代理对象解析为AddressSchema，从而抛出类型错误。另外，如果创建Event时直接传入字典格式的address数据，SQLAlchemy无法自动转换为Address实例，也会导致后续解析失败。

解决方法 #

方法1：查询时预加载关联数据 #

查询Event实例时，使用joinedload强制预加载address关联数据：

from sqlalchemy.orm import joinedload

# 查询时预加载address
event_instance = db_session.query(Event).options(joinedload(Event.address)).get(event_id)
# 此时可正常解析
schema = EventSchema.from_orm(event_instance)
print(schema.json())

方法2：修改关系为立即加载 #

在SQLAlchemy模型定义中，将关系的lazy参数设为'joined'或'eager'，让关联数据在查询主模型时自动加载：

class Event(Base):
    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    address = relationship(
        'Address',
        uselist=False,
        back_populates='event',
        lazy='joined'  # 改为立即加载
    )

方法3：创建实例时正确关联Address对象 #

如果通过Event(**data)创建实例，需先将address字典转换为Address实例：

# 从data中取出address字段并转换为实例
address_data = data.pop('address')
address = Address(**address_data)
event = Event(**data, address=address)
db_session.add(event)
db_session.commit()

# 会话未关闭时address已加载，可直接解析
schema = EventSchema.from_orm(event)

方法4：开启Pydantic ORM模式支持 #

确保Pydantic Schema开启ORM模式，才能正确解析SQLAlchemy实例：

• Pydantic v1：添加orm_mode = True到Config类
• Pydantic v2：添加from_attributes = True到Config类

# Pydantic v2示例
class EventSchema(BaseModel):
    id: int
    address: Optional[AddressSchema] = Field(...)
    
    class Config:
        from_attributes = True

最终效果 #

按上述方法处理后，调用EventSchema.from_orm(event_instance)会输出符合预期的JSON格式：

{"id": 1, "address": {"country": "foocountry", "city": "foocity", "street": "foostreet", "build": "foobuild"}}

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