如何使用 SQLAlchemy、FastAPI、Pydantic、Postgres 和 Mapped[List[Obj]] 字段查询具有多对多关系的表？

Question

我已经这样做了 4 天，我真的需要一些帮助。 正如标题中所讨论的，我正在使用 Python、SQLAlchemy、PostgreSQL、FastAPI 和 Pydantic（除其他外）。 我正在取回元组，但我不明白发生了什么。 另外，我现在才看到它是元组中的 repr...嗯。

我已经为 SQLAlchemy 定义了 Buyer、HSCode 和 BuyerHSCode 的模型。

class Base(DeclarativeBase, AsyncAttrs):
    pass


class TimestampMixin(object):
    @declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

    created_at = mapped_column(DateTime, nullable=False, server_default=func.now())
    updated_at = mapped_column(
        DateTime, nullable=False, server_default=func.now(), onupdate=func.now()
    )

class Buyer(Base, TimestampMixin):
    __tablename__ = "buyers"

    uid: Mapped[uuid.UUID] = mapped_column(
        pg.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, nullable=False
    )
    company_name: Mapped[str] = mapped_column(String(100), nullable=False)
    region: Mapped[str] = mapped_column(String(15), nullable=False)
    country: Mapped[str] = mapped_column(String(56), nullable=False)
    city: Mapped[str] = mapped_column(String(100), nullable=False)
    company_size: Mapped[str] = mapped_column(String(30), nullable=False)

    hs_codes: Mapped[List["HSCode"]] = relationship("HSCode", secondary='buyer_hs_codes', back_populates="buyers")

    def __repr__(self):
        return f"<Buyer(company_name='{self.company_name}')>"

class HSCode(Base):
    __tablename__ = "hs_codes"

    uid: Mapped[uuid.UUID] = mapped_column(
        pg.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, nullable=False
    )
    hs_code: Mapped[str] = mapped_column(String(12), nullable=False)

    buyers: Mapped[List["Buyer"]] = relationship("Buyer", secondary='buyer_hs_codes', back_populates="hs_codes")

class BuyerHSCode(Base):
    __tablename__ = "buyer_hs_codes"

    buyer_uid: Mapped[uuid.UUID] = mapped_column(
        pg.UUID(as_uuid=True), ForeignKey('buyers.uid'), primary_key=True
    )
    hs_code_uid: Mapped[uuid.UUID] = mapped_column(
        pg.UUID(as_uuid=True), ForeignKey('hs_codes.uid'), primary_key=True
    )

    def __repr__(self):
        return f"<BuyerHSCode(buyer_id='{self.buyer_uid}', hs_code_id='{self.hs_code_uid}')>"

我创建了多个类来以 Python 方式表示数据。

class BuyerResponseModel(BaseModel):
    company_name: str
    region: str
    country: str
    city: str
    company_size: str
    hs_codes: List[HSCodeResponse]
    created_at: str
    updated_at: str

    class Config:
        from_attributes = True

class HSCodeResponse(BaseModel):
    hs_code: str

我创建了 FastAPI 端点以及一个类来表示买方端点的所有服务。

@buyer_router.get("/", response_model=List[BuyerResponseModel])
async def get_all_buyers_endpoint(session: AsyncSession = Depends(get_session)):
    return await buyer_service.get_all_buyers(session)

async def get_all_buyers(self, session: AsyncSession):
    statement = (
        select(Buyer)
        .options(selectinload(Buyer.hs_codes))
        .order_by(desc(Buyer.created_at))
    )

    result = await session.exec(statement)
    buyers = result.all()

        return buyers

使用 FastAPI 文档时出错：

fastapi.exceptions.ResponseValidationError: 8 validation errors:
  {'type': 'missing', 'loc': ('response', 0, 'company_name'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'region'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'country'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'city'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'company_size'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'hs_codes'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'created_at'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}
  {'type': 'missing', 'loc': ('response', 0, 'updated_at'), 'msg': 'Field required', 'input': (<Buyer(company_name='company1')>,)}

终端输出：

2024-10-18 10:31:54,924 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2024-10-18 10:31:54,931 INFO sqlalchemy.engine.Engine SELECT buyers.uid, buyers.company_name, buyers.region, buyers.country, buyers.city, buyers.company_size, buyers.created_at, buyers.updated_at 
FROM buyers ORDER BY buyers.created_at DESC
2024-10-18 10:31:54,931 INFO sqlalchemy.engine.Engine [generated in 0.00051s] ()
2024-10-18 10:31:54,952 INFO sqlalchemy.engine.Engine SELECT buyers_1.uid AS buyers_1_uid, hs_codes.uid AS hs_codes_uid, hs_codes.hs_code AS hs_codes_hs_code 
FROM buyers AS buyers_1 JOIN buyer_hs_codes AS buyer_hs_codes_1 ON buyers_1.uid = buyer_hs_codes_1.buyer_uid JOIN hs_codes ON hs_codes.uid = buyer_hs_codes_1.hs_code_uid 
WHERE buyers_1.uid IN ($1::UUID)
2024-10-18 10:31:54,952 INFO sqlalchemy.engine.Engine [generated in 0.00045s] (UUID('70bd135d-841c-49c6-997d-da18d3b34e4a'),)
2024-10-18 10:31:54,966 INFO sqlalchemy.engine.Engine ROLLBACK
INFO:     127.0.0.1:50269 - "GET /api/v1/buyers/ HTTP/1.1" 500 Internal Server Error

这是 FastAPI 文档对端点的期望...

[
  {
    "company_name": "string",
    "region": "string",
    "country": "string",
    "city": "string",
    "company_size": "string",
    "hs_codes": [
      {
        "hs_code": "string"
      }
    ],
    "created_at": "string",
    "updated_at": "string"
  }
]

如有任何问题，请随时提出以进行澄清。 我是 FastAPI、SQLAlchemy 和 PostgreSQL 的新手。 我通常使用 Flask 并自己创建所有这些，但我完全不知所措，确实需要一些帮助。 谢谢你。

Answer 1

SQLAlchemy 将返回一个

Row

对象列表，在本例中本质上是长度为 1 的元组。要使其仅返回

Buyer

列表，最简单的方法是将

session.exec(statement)

替换为

session.scalars(statement)

。