如何使用 Pydantic/FastAPI 正确处理 SQLAlchemy 的连接结果

Question

我想要一条关于处理在 SQLAlchemy 中执行的连接操作的结果并使用 Pydantic（在 FastAPI 中）进行序列化的建议。

如果我没记错的话，两个表的连接结果会导致 SQLAlchemy 模型的元组列表。这是一个模拟，like_a_join 是我对连接查询结果的理解。

from pydantic import BaseModel
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class A(BaseModel):
    a: int

    class Config:
        orm_mode = True


class B(BaseModel):
    b: int

    class Config:
        orm_mode = True


class Am(Base):
    __tablename__ = "A"
    a = Column(Integer, primary_key=True, nullable=False)


class Bm(Base):
    __tablename__ = "B"
    b = Column(Integer, primary_key=True, nullable=False)


def like_a_join():
    return [(Am(a=1), Bm(b=1))]

虽然可以使用 from_orm 将模型对象传递给 Pydantic 以进行简单查询（例如在 FastAPI Postgres cookie 切割器 here 上完成，但这对我来说如何最好地处理连接并不明显/元组大小写。

创建一个类来处理如下的元组有意义吗？

from typing import List
from pydantic import parse_obj_as
from dataclasses import dataclass

@dataclass
class Cm:
    a: Am
    b: Bm

class C(BaseModel):
    a: A
    b: B

    class Config:
        orm_mode = True

def like_a_join_with_class():
    return [Cm(a=Am(a=1), b=Bm(b=1))]

print(parse_obj_as(List[C], like_a_join_with_class()))

用字典会更好吗？

def like_a_join_with_dict():
    return [{"a": Am(a=1), "b": Bm(b=1)}]

背后的想法是将查询结果包含在 FastAPI 端点中，并自动处理序列化。

提前感谢您的帮助。

Answer 1

拥有一个 pydantic 模型来处理和操作您的数据总是好的，但是当您想从端点返回这样的连接时，它尤其实用。在这种情况下，您可以使用 pydantic 模型来覆盖路径操作的 response_model 参数：

@router.get("/some_endpoint_path", response_model=SomePydanticModel)
def request_handler():
    ...

如果这是您的情况，那么我将创建一个涵盖整个查询结果的模型（即 2 元组 [A，B] 的列表）。如果 ORM Am 映射到 pydantic A，并且 ORM Bm 映射到 pydantic B，那么整个查询结果应该成功映射到以下 pydantic 模型：

import typing as t

class JoinResult(BaseModel):
    results: t.List[t.Tuple[A, B]]

    class Config:
        orm_mode = True

sqlalchemy 查询的确切形式取决于您使用的 sqlalchemy 版本和您选择的抽象级别，但是对于具有异步 ORM 会话的现代 2.x 样式查询，它看起来（或多或少）如下所示：

statement = select(Am, Bm).join(<your_join_here>).where(<your_condition_here>)
result = await session.execute(statement)
scalar_results = result.scalars()

这个scalar_results 应该是类似列表的对象，其中包含两个元组，其中包含 sqlalchemy 模型 Am 和 Bm 的实例。您应该能够按如下方式解析它：

jr = JoinResult(results=iter(scalar_results))

然后你可以直接从你的路径操作函数中返回jr。

或者，如果您真的想直接返回查询结果， 您可以尝试来自 tiangolo 的新项目：SQLModel。这是一个 pydantic 和 sqlalchemy 之间缺少桥梁。它可能会给您带来一些想法和解决方案。但请记住，SQLModel 处于开发的早期阶段，可能存在许多错误。不过举报他们，tiangolo 会很高兴的 :)