如何结合SQLAlchemy AsyncAttrs与Pydantic的.from_obj方法？

要让ASchema.from_orm(a)（Pydantic中处理ORM对象的标准方法）自动处理SQLAlchemy AsyncAttrs的异步关联属性，无需手动await，可以通过以下两种实用方式实现：

方案1：自定义异步类方法（最直观） #

在ASchema中封装异步加载关联属性的逻辑，直接提供一个异步入口方法：

from pydantic import BaseModel

class ASchema(BaseModel):
    id: int
    bs_id: int
    bs: BsRelationSchema

    class Config:
        from_attributes = True

    @classmethod
    async def from_async_obj(cls, db_obj):
        # 预加载异步关联属性，加载后db_obj.bs会被填充
        await db_obj.awaitable_attrs.bs
        # 调用原生from_orm完成转换
        return cls.from_orm(db_obj)

使用方式： #

# 直接调用异步方法，无需手动处理关联加载
a_schema = await ASchema.from_async_obj(a)

方案2：异步字段序列化器（自动处理单个字段） #

利用Pydantic V2的异步字段序列化器，在序列化bs字段时自动完成异步加载：

from pydantic import BaseModel, field_serializer

class ASchema(BaseModel):
    id: int
    bs_id: int
    bs: BsRelationSchema

    class Config:
        from_attributes = True

    @field_serializer("bs", is_async=True)
    async def serialize_bs(self, _, info):
        # 从context中获取原ORM对象，需提前传入
        db_obj = info.context.get("db_obj")
        if db_obj:
            loaded_bs = await db_obj.awaitable_attrs.bs
            return BsRelationSchema.from_orm(loaded_bs)
        return None

使用方式： #

# 传入context传递原ORM对象，调用异步验证方法完成转换
a_schema = await ASchema.model_validate_async(a, context={"db_obj": a})

注意事项 #

• 若使用Pydantic V1，异步序列化器不被支持，优先选择方案1。
• 若模型存在多个异步关联属性，只需在from_async_obj中添加对应await语句即可，比如await db_obj.awaitable_attrs.another_relation。

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