如何使用 SQLAlchemy 和 Pydantic 将项目计数添加到 FastAPI API？答案

【问题标题】：How to add count of items to a FastAPI API using SQLAlchemy and Pydantic?如何使用 SQLAlchemy 和 Pydantic 将项目计数添加到 FastAPI API？
【发布时间】：2021-12-31 18:13:43
【问题描述】：

我有三个 SQLAlchemy 模型，比如说一个“项目”表、一个“工人”表和一个“设备”表。我正在使用 Pydantic Schemas。

工人和设备可以通过外键分配给项目（即worker.project_id ==project.id）。

在我的项目 API 端点中，除了项目信息之外，我还想给出分配给项目的工人和设备的数量（计数）。

如果我的“项目”表中没有用于这些计数的列，因此我的模型中没有用于传递它们的列，我该怎么做？

class project(Base):
    __tablename__ = "projects"
    id = Column(Integer, primary_key=True)
    name = Column(String(80), nullable=False)
    site_name = Column(String(80), nullable=False)
    company_id = Column(Integer, ForeignKey('companies.id'), nullable = False)


class worker(Base):
    __tablename__ = "workers"
    id = Column(Integer, primary_key=True)
    fullname = Column(String(80), nullable=False)
    project_id = Column(Integer, ForeignKey('projects.id')


class equipment(Base):
    __tablename__ = "equipments"
    id = Column(Integer, primary_key=True)
    name = Column(String(80), nullable=False)
    project_id = Column(Integer, ForeignKey('projects.id')

---------------------------------------------------------------
router = APIRouter(tags=['Projects'])

@router.get('/projects/', response_model=List[ProjectSchema])
def all_projects(db:Session=Depends(get_db)):
    return project_fn.get_all(db)

-----------------------------------------------
project_fn.py

def get_all(db:Session=Depends(get_db)):
    projects = db.query(project).all()
    return projects

-----------------------------------------------
class ProjectSchema(BaseModel):
    id: int
    name: str
    site_name: str = Field(min_length=3)
    company: CompanyBase

    class Config:
        orm_mode = True

----------------------------------------------
Sample API I need:

{
    "id": 1,
    "name": "Project-01",
    "site_name": "Site-01",
    "company": {
      "id": 1,
      "name": "Company-01",
    }
    "worker_count" : 5,
    "equipment_count" : 3
}

如何在此处获取 'worker_count' 和 'equipment_count'？

【问题讨论】：

您可以使用func.count获取相关列的分组计数；您可能需要稍微修改数据以获得所需的结构（即，而不是 (Project, WorkerCount, EquipmentCount)，因为您想要 project 对象中的所有内容）。请参阅stackoverflow.com/questions/25500904/… 了解如何执行 SQLAlchemy 部分。
@MatsLindh 感谢您的反馈，但我仍然不清楚该怎么做。我可以在我的“项目”模型中使用func.count 来查询其他表吗？我可以通过“工人”或“设备”模型中的查询轻松获得计数，但主要问题是如何将它们传递给“项目”API。如您提到的链接中使用的那样，在此处创建辅助表是否有帮助？因为显然我不能将 'worker_id' 和 'equipment_id' 添加到我的 'project' 表中，因为这些是多对多的关系。
你没有包含project_fn，所以很难说你是如何查询数据库的，但是session.query(project, func.count(worker.id), func.count(equipment.id)).join(worker, isouter=True).join(equipment, isouter=True).all() 可以工作，我认为它会给你一个包含三个条目的元组；项目、工人数和设备数
@MatsLindh 我添加了 project_fn 和 Pydantic 模式。也许我在这里遗漏了一些明显的东西，但我假设当我们坚持 SQLAlchemy 和 Pydantic 模型时，我们只能将这些模型中的现有字段传递给我们的 API。在 SQLAlchemy 中，我假设我们只能将“列”或“关系”作为字段，而不是任何其他自定义字段？我在吗？
您是否尝试修改您的查询以包括我在评论中包含的两个函数以及显式连接？它应该给你一个包含项目和计数的三元素元组。

标签： python api sqlalchemy fastapi pydantic

【解决方案1】：

好的，我找到了下面的答案并使用column_property访问其他表并计算项目模型中所需的项目：

class project(Base):
    __tablename__ = "projects"
    id = Column(Integer, primary_key=True)
    name = Column(String(80), nullable=False)
    site_name = Column(String(80), nullable=False)
    company_id = Column(Integer, ForeignKey('companies.id'), nullable = False)

    worker_count = column_property(select([func.count(worker.id)]).filter(worker.project_id==id).scalar_subquery())

    equipment_count = column_property(select([func.count(equipment.id)]).filter(equipment.project_id==id).scalar_subquery())

【讨论】：