python(fastapi) 实现更新，新增，删除接口

1.新增#新增接口#用户输入图书信息ID,书名。作者价格出版社#用户输入参数》请求体参数classBookCreate(BaseModel):id:intbookname:strauthor:strprice:floatpublisher:strapp.post(/book/add)asyncdefadd_book(book:BookCreate,db:AsyncSessionDepends(get_database)):#orm对象 转new_bookBook(**book.model_dump())db.add(new_book)awaitdb.commit()return{message:添加成功}接口测试2.更新#更新接口classBookUpdate(BaseModel):bookname:strauthor:strprice:floatpublisher:strapp.put(/book/update/{book_id})asyncdefupdate_book(book_id:int,data:BookUpdate,db:AsyncSessionDepends(get_database)):#1.查找图书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.publisher# 提交修改awaitdb.commit()return{message:更新成功}测试接口3.删除#删除接口app.delete(/book/delete/{book_id})asyncdefdelete_book(book_id:int,db:AsyncSessionDepends(get_database)):#先查在删 提交db_bookawaitdb.get(Book,book_id)ifdb_bookisNone:raiseHTTPException(status_code404,detail图书未找到)awaitdb.delete(db_book)awaitdb.commit()return{message:删除成功}测试接口完整代码fromunittestimportresultfromfastapiimportFastAPI,Depends,HTTPExceptionfromdatetimeimportdatetimefromsqlalchemyimportString,DateTime,func,selectfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfrompydanticimportBaseModel appFastAPI()# 1.创建异步引擎#ASYNC_DATAASE_URL为数据库的地址ASYNC_DATAASE_URLmysqlaiomysql://root:630229localhost:3306/fast_api_db?charsetutf8async_enginecreate_async_engine(ASYNC_DATAASE_URL,echoTrue,# 可选删除SQL日志pool_size10,#设置连接池活跃的连接数max_overflow20,# 允许额外的连接数)# 2.定义模型类: 基类表对应的模型类# 基类:创建时间、更新时间;书籍表:id、书名、作者、价格、出版社classBase(DeclarativeBase):create_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultdatetime.now,comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),onupdatefunc.now(),defaultdatetime.now,comment修改时间)classBook(Base):__tablename__bookid:Mapped[int]mapped_column(primary_keyTrue,comment书籍id)bookname:Mapped[str]mapped_column(String(255),comment书名)author:Mapped[str]mapped_column(String(255),comment作者)price:Mapped[float]mapped_column(comment价格)publisher:Mapped[str]mapped_column(String(255),comment出版社)# 3.启动应用时建表FASTAPI启动时调用建表的函数asyncdefcreate_tables():#创建异步引擎创建事务建表asyncwithasync_engine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)#模型类的元数据app.on_event(startup)asyncdefstartup_event():awaitcreate_tables()app.get(/)asyncdefroot():return{message:Hello World}# 路由匹配中使用 ORM ,创建依赖项使用 Depends注入到处理函数# 需求查询功能的接口查询图书》依赖注入创建依赖项获取数据库会话Depends注入路由处理函数# 4.创建异步会话工厂AsyncSessionLocalasync_sessionmaker(bindasync_engine,# 绑定异步引擎class_AsyncSession,# 设置会话类为异步会话expire_on_commitFalse# 设置会话提交时是否自动刷新)# 5.依赖项获取数据库会话asyncdefget_database():asyncwithAsyncSessionLocal()assession:try:yieldsession# 返回数据库会话awaitsession.commit()# 提交事务exceptException:awaitsession.rollback()# 回滚事务raisefinally:awaitsession.close()# 关闭会话# 6.路由查询图书获取所有app.get(/book/books)asyncdefget_books_list(session:AsyncSessionDepends(get_database)):resultawaitsession.execute(select(Book))#查询返回一个ORM对象booksresult.scalars().all()#获取所有returnbooks# 7.路由查询图书获取第一个app.get(/book/firstbook)asyncdefget_books_list(session:AsyncSessionDepends(get_database)):resultawaitsession.execute(select(Book))#查询返回一个ORM对象bookresult.scalars().first()#获取第一个returnbook# 7.路由查询图书获取单条数据根据ID查询app.get(/book/IDbook)asyncdefget_books_list(session:AsyncSessionDepends(get_database)):bookawaitsession.get(Book,3)#获取单条数据根据ID查询returnbook# 8.需求路径参数书籍idapp.get(/book/get_book/{book_id})asyncdefget_book(book_id:int,session:AsyncSessionDepends(get_database)):resawaitsession.execute(select(Book).where(Book.idbook_id))bookres.scalar_one_or_none()returnbook# 9.需求条件查询价格大于等于app.get(/book/search_book)asyncdefget_books_list(book_price:int,session:AsyncSessionDepends(get_database)):resawaitsession.execute(select(Book).where(Book.pricebook_price))booksres.scalars().all()returnbooks#10 模糊查询app.get(/book/like_book)asyncdefget_books_list(session:AsyncSessionDepends(get_database)):#需求 作者author以周开头查询 % _#like() 模糊查询% 任意个字符; _一个单个字符#res await session.execute(select(Book).where(Book.author.like(周%)))#res await session.execute(select(Book).where(Book.author.like(周_)))# | ~ 逻辑运算符# res await session.execute(select(Book).where((Book.author.like(周%)) % (Book.price30)))# res await session.execute(select(Book).where((Book.author.like(周%)) | (Book.price 30)))#in() 包含resawaitsession.execute(select(Book).where(Book.author.in_([周树人,周国平])))booksres.scalars().all()returnbooks#聚合查询app.get(/book/count)asyncdefget_book_count(db:AsyncSessionDepends(get_database)):# 聚合查询 select(func.方法名模型类.属性#count统计行数量#result await db.execute(select(func.count(Book.id)))#max求最大值#result await db.execute(select(func.max(Book.price)))#min求最小值resultawaitdb.execute(select(func.min(Book.price)))#avg求平均值# result await db.execute(select(func.avg(Book.price)))#sum求和# result await db.execute(select(func.sum(Book.price)))# scalar() 提取单个值空表时返回 0countresult.scalar()return{count:count}#分页查询app.get(/book/page)asyncdefget_book_page(page:int1,page_size:int2,db:AsyncSessionDepends(get_database)):resultawaitdb.execute(select(Book).offset((page-1)*page_size).limit(page_size))booksresult.scalars().all()returnbooks#新增接口#用户输入图书信息ID,书名。作者价格出版社#用户输入参数》请求体参数classBookCreate(BaseModel):id:intbookname:strauthor:strprice:floatpublisher:strapp.post(/book/add)asyncdefadd_book(book:BookCreate,db:AsyncSessionDepends(get_database)):#orm对象 转new_bookBook(**book.model_dump())db.add(new_book)awaitdb.commit()return{message:添加成功}#更新接口classBookUpdate(BaseModel):bookname:strauthor:strprice:floatpublisher:strapp.put(/book/update/{book_id})asyncdefupdate_book(book_id:int,data:BookUpdate,db:AsyncSessionDepends(get_database)):#1.查找图书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.publisher# 提交修改awaitdb.commit()return{message:更新成功}#删除接口app.delete(/book/delete/{book_id})asyncdefdelete_book(book_id:int,db:AsyncSessionDepends(get_database)):#先查在删 提交db_bookawaitdb.get(Book,book_id)ifdb_bookisNone:raiseHTTPException(status_code404,detail图书未找到)awaitdb.delete(db_book)awaitdb.commit()return{message:删除成功}