Python Web Day05聚合查询func.方法模型类.属性方法名作用count统计行数量avg求平均值max求最大值min求最小值sum求和FastAPI 的路由匹配和代码定义顺序有关。固定路径例如/books/count应该写在动态路径/books/{book_id}前面否则固定路径中的字符串可能会被误识别为路径参数导致类型校验失败。案例app.get(/books/count)asyncdefget_book_count(db:AsyncSessionDepends(get_datebase)):resultawaitdb.execute(select(func.count(Book.id)))numresult.scalar()returnnumapp.get(/books/{book_id})asyncdefget_book_list(book_id:int,db:AsyncSessionDepends(get_datebase)):resultawaitdb.execute(select(Book).where(Book.idbook_id))bookresult.scalar_one_or_none()returnbook如果 app.get(“/books/count”)放在app.get(“/books/{book_id}”)前的话count有可能被当作book_id,导致报错。分页查询 select().offset().limit() offset:跳过的记录数 limit返回的记录数app.get(/book/get_book_list)asyncdefget_book_list(db:AsyncSessionDepends(get_datebase),page:int1,page_size:int1,):skip(page-1)*page_size stmtselect(Book).offset(skip).limit(page_size)resultawaitdb.execute(stmt)booksresult.scalars().all()returnbooksapp.get(/book/get_book_list)asyncdefget_book_list(db:AsyncSessionDepends(get_datebase),page:int1,page_size:int1,):skip(page-1)*page_size stmtselect(Book).offset(skip).limit(page_size)resultawaitdb.execute(stmt)booksresult.scalars().all()returnbooks查询总结写法含义scalar()取第一行第一列的一个值scalars()把每一行的第一列都提取出来scalars().all()提取所有 ORM 对象并转成列表scalars().first()提取第一个数据scalars_one_or_none()提取一个或NULLscalars()是 SQLAlchemy 查询结果处理中非常常用的方法增加1.定义ORM对象 2.添加对象到事务add对象3.commit提交到数据库ORM对象数据库表里某一行数据在 Python 代码里的对象形式。它的作用是把数据库中的表、字段、记录映射成 Python 里的类、属性、对象。classBookBase(BaseModel):id:intbookname:strauthor:strprice:floatpublisher:strapp.post(/book/add)asyncdefadd_book(book:BookBase,db:AsyncSessionDepends(get_datebase)):# ORM对象 add commitbook_objBook(**book.__dict__)db.add(book_obj)awaitdb.commit()returnbook反思这个是Pydantic 模型类主要用于接口数据校验。classBookBase(BaseModel):id:intbookname:strauthor:strprice:floatpublisher:str这个是SQLAlchemy ORM 模型类主要用于数据库操作class Base(DeclarativeBase): create_time:Mapped[datetime] mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now,comment创建时间) update_time:Mapped[datetime] mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now,onupdatefunc.now,comment修改时间) class Book(Base): __tablename__ book id:Mapped[int] mapped_column(primary_keyTrue,comment书籍) bookname:Mapped[str] mapped_column(String(255),comment书名) author: Mapped[str] mapped_column(String(255),comment作者) price:Mapped[float] mapped_column(Float,comment价格) publisher:Mapped[str] mapped_column(String(255),comment出版社)它们主要是通过字段名对应起来的。__tablename__book上面是SQLAlchemy ORM 模型里的配置用来指定这个类对应数据库中的哪张表。更新查询 属性重新赋值 commit提交到数据库classBookUpdate(BaseModel):bookname:strauthor:strprice:floatpublisher:strapp.put(/books/update/{book_id})asyncdefupdate_book(book_id:int,data:BookUpdate,db:AsyncSessionDepends(get_datebase)):db_bookawaitdb.get(Book,book_id)ifdb_bookisNone:raiseHTTPException(status_code404,detail查无此书)db_book.booknamedata.bookname db_book.authordata.author db_book.pricedata.price db_book.publisherdata.publisherawaitdb.commit()awaitdb.refresh(db_book)return修改成功遇到的问题classBase(DeclarativeBase):create_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now(),comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now(),onupdatefunc.now(),comment修改时间)1.func.now()要加2.update_book函数和get_datebase只需要一个提交await db.commit()删除查询get delete删除 commit提交到数据库app.put(/books/delete/{book_id})asyncdefdelete_book(book_id:int,db:AsyncSessionDepends(get_datebase)):db_bookawaitdb.get(Book,book_id)ifdb_bookisNone:raiseHTTPException(status_code404,detail无)awaitdb.delete(db_book)awaitdb.commit()return修改成功