如何将带有计数结果的列添加到 sqlalchemy 查询并通过 pydantic 检查？

Question

描述

以前，我的查询返回单个故事 table 的内容。现在我想添加更多信息：我需要为每个 Story 输出 prizes_count。 Stories table 中没有字段 prizes_count 所以我进行了以下查询。

db.query(models.Stories, func.count(models.Stories.prizes).label("prizes_count")).join(models.Prizes)\
        .group_by(models.Stories.id).all()

但是我有两个问题。

我从 Pydantic 收到验证错误，因为此查询 returns 元组列表，如 (<database.models.Stories object at 0x0000026BB0055E20>, 1)。我必须将 prizes_count 值插入 Stories 对象，反之亦然，将所有字段拉入元组。当然，我可以手动完成，但我认为有更好的方法。
通过此查询，我丢失了所有奖品为 0 的故事，因为我的加入忽略了它们。

代码

端点

@app.get("/stories/", response_model=List[schemas.StoryFullInfo])
def get_stories(db: Session = Depends(get_db)):
    return crud.get_stories(db)

crud

def get_stories(db: Session):
    return db.query(models.Stories, func.count(models.Stories.prizes).label("prizes_count")).join(models.Prizes)\
        .group_by(models.Stories.id).all()

型号

class Stories(Base):
    __tablename__ = "stories"
    id = Column(INTEGER(unsigned=True), primary_key=True)
    title = Column(String(length=128), index=True)
    text = Column(String(length=1000))
    author_id = Column(INTEGER(unsigned=True), ForeignKey("users.id", onupdate="CASCADE", ondelete="CASCADE"),
                       nullable=False)
    status = Column(TINYINT(unsigned=True), server_default="0")
    genre_type = Column(TINYINT(unsigned=True), server_default="0")
    likes_count = Column(INTEGER(unsigned=True), server_default="0")
    image = Column(Text)
    added_to_best_by = Column(INTEGER(unsigned=True))
    creation_DT = Column(DateTime, server_default=func.now())
    change_status_DT = Column(DateTime)

    author = relationship("Users", back_populates="stories")
    comments = relationship("Comments", back_populates="story")
    prizes = relationship("Prizes", back_populates="story")


class Prizes(Base):
    __tablename__ = "prizes"
    id = Column(INTEGER(unsigned=True), primary_key=True)
    title = Column(String(length=128), nullable=False)
    image_id = Column(TINYINT(unsigned=True))
    story_id = Column(INTEGER(unsigned=True), ForeignKey("stories.id", onupdate="CASCADE", ondelete="CASCADE"),
                      nullable=False)
    user_id = Column(INTEGER(unsigned=True), ForeignKey("users.id", onupdate="CASCADE", ondelete="CASCADE"),
                     nullable=False)
    text = Column(String(length=512), nullable=False)
    creation_DT = Column(DateTime, server_default=func.now())

    story = relationship("Stories", back_populates="prizes")
    author = relationship("Users", back_populates="prizes")

架构

class StoryBaseInfo(BaseModel):
    id: int
    title: str = None
    author_id: int

    class Config:
        orm_mode = True


class StoryUpdateInfo(StoryBaseInfo):
    #title: str = None
    text: str = None
    status: int
    genre_type: int
    likes_count: int
    image: str = None
    added_to_best_by: int = None
    change_status_DT: datetime = None

    class Config:
        orm_mode = True


class StoryFullInfo(StoryUpdateInfo):

    creation_DT: datetime
    author: UserBaseInfo
    prizes_count: int

    class Config:
        orm_mode = True


class PrizeBaseInfo(BaseModel):
    id: int
    story_id: int

    class Config:
        orm_mode = True


class PrizeInfo(PrizeBaseInfo):
    title: str
    image_id: int
    text: str
    creation_DT: datetime

    author: UserBaseInfo
    story: StoryBaseInfo

    class Config:
        orm_mode = True

Answer 1

嗯，原来我问这个的时候想错了方向。 SQLAlchemy 的特性解决了这个问题。我可以使用我配置的关系来计算奖品。我的解决方案是将混合属性添加到 Stories SQLAlchemy 模型

class Stories(Base):
    __tablename__ = "stories"
    id = Column(INTEGER(unsigned=True), primary_key=True)
    title = Column(String(length=128), index=True)
    text = Column(String(length=1000))
    author_id = Column(INTEGER(unsigned=True), ForeignKey("users.id", onupdate="CASCADE", ondelete="CASCADE"),
                       nullable=False)
    status = Column(TINYINT(unsigned=True), server_default="0")
    genre_type = Column(TINYINT(unsigned=True), server_default="0")
    likes_count = Column(INTEGER(unsigned=True), server_default="0")
    image = Column(Text)
    added_to_best_by = Column(INTEGER(unsigned=True))
    creation_DT = Column(DateTime, server_default=func.now())
    change_status_DT = Column(DateTime)

    author = relationship("Users", back_populates="stories")
    comments = relationship("Comments", back_populates="story")
    prizes = relationship("Prizes", back_populates="story")

    @hybrid_property
    def prizes_count(self):
        return len(self.prizes)

然后下面的查询将满足 Pydantic 方案。

def get_stories(db: Session):
    return db.query(models.Stories).all()

如何将带有计数结果的列添加到 sqlalchemy 查询并通过 pydantic 检查？

How to add a column with a count result to a sqlalchemy query and pass the pydantic check?

sqlalchemy

pydantic

fastapi

描述

代码