如何使用 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))]

虽然可以将模型对象传递给 Pydantic 以进行from_orm简单查询（例如在 FastAPI Postgres 饼干切割机上完成），但对我来说，如何最好地处理连接/元组情况并不明显。

创建一个类来处理如下所示的元组是否有意义？

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] 的列表）。如果 ORMAm映射到 pydantic A，并且 ORMBm映射到 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和2 元组Bm。您应该能够按如下方式解析它：

jr = JoinResult(results=iter(scalar_results))

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

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