python（Fastapi）实现查询功能（条件查询，模糊查询，与非查询，包含查询）

表数据1.查询测试接口2.条件查询比较判断查询测试接口3.模糊查询 与非查询包含查询完整代码fromunittestimportresultfromfastapiimportFastAPI,DependsfromdatetimeimportdatetimefromsqlalchemyimportString,DateTime,func,selectfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_column 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