文章目录前言一、装包指令二、main.py示例代码前言学习Python语法做的练习代码记录同时也供大家参考相互学习一、装包指令pip install “sqlalchemy[asyncio]” aiomysql二、main.py示例代码fromsqlalchemyimportString,Float,DateTime,select,funcfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfromdatetimeimportdatetimeimportasynciofromsqlalchemy.sql.elementsimportand_ 初始化异步引擎 # 1、定义连接字符串connectionStringmysqlaiomysql://root:xxxlocalhost:3306/xxx?charsetutf8# 2、创建异步引擎asyncEnginecreate_async_engine(connectionString,echoTrue,pool_size10,max_overflow20)# 3、声明实体模型类 模型基类 classMyBaseModel(DeclarativeBase):CreateDatetime:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now(),commentCreateDate)classBook(MyBaseModel):__tablename__bookId:Mapped[str]mapped_column(String(50),primary_keyTrue,commentID)Title:Mapped[str]mapped_column(String(50),nullableTrue,commentTitle)Author:Mapped[str]mapped_column(String(20),nullableTrue,commentAuthor)Price:Mapped[float]mapped_column(Float,nullableTrue,commentPrice)# 4、初始数据库asyncdefinitDatabase():asyncwithasyncEngine.begin()asconn:awaitconn.run_sync(MyBaseModel.metadata.create_all)# 5、创建异步会话工厂AsyncSessionLocalasync_sessionmaker(bindasyncEngine,class_AsyncSession,expire_on_commitFalse)# 6、获取数据库会话asyncdefgetDatabase():asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()exceptException:awaitsession.rollback()finally:awaitsession.close() 7、 实体模型操作 # 获取所有书籍信息asyncdefqueryBooks():asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book))returnresult.scalars().all()# 获取第一本书asyncdefqueryFirstBook():asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book))returnresult.scalars().first()# 根据id获取书籍asyncdefqueryBookById(id):asyncwithAsyncSessionLocal()assession:returnawaitsession.get(Book,id)# 条件筛选查询# 比较判断 ;;;;# 模糊匹配 like() Book.Title.like(西游%)# 与非查询 ;|;~ (Book.Author曹雪芹)(Book.Price50)# 包含查询 in_()asyncdefqueryBooksByPrice(price):asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book).where(Book.Price50))returnresult.scalars().all()# 精准匹配 singeOrDefaultasyncdefqueryBookByTitle(title):asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book).where(Book.Titletitle))returnresult.scalar_one_or_none()# 批量筛选书籍asyncdefbatchQueryBooks(ids):asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book).where(Book.Id.in_(ids)))returnresult.scalars().all()# 聚合操作asyncdefjointOperation():asyncwithAsyncSessionLocal()assession:# count# resultawait session.execute(select(func.count(Book.Id)))# avg# resultawait session.execute(select(func.avg(Book.Price)))# max/min# resultawait session.execute(select(func.max(Book.Price)))# sumresultawaitsession.execute(select(func.sum(Book.Price)))returnresult.scalar()# 聚合操作带条件筛选asyncdefjointOperationV2():asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(func.sum(Book.Price)).where(Book.Price50))returnresult.scalar()# 分页查询asyncdefqueryPagingDatas(pageIndex1,pageSize3):asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book).offset((pageIndex-1)*pageSize).limit(pageSize))returnresult.scalars().all()# 分页查询筛选条件自定义排序字段asyncdefqueryPagingDatasV2(minPrice,pageIndex1,pageSize3):asyncwithAsyncSessionLocal()assession:resultawaitsession.execute(select(Book).where(Book.PriceminPrice).order_by(Book.CreateDatetime.desc())#asc().offset((pageIndex-1)*pageSize).limit(pageSize))returnresult.scalars().all()# 分页查询筛选条件延迟执行classQueryPagingDatasV3SearchInfo:TitleNonePriceNoneasyncdefqueryPagingDatasV3(searchInfo:QueryPagingDatasV3SearchInfo,pageIndex1,pageSize3):asyncwithAsyncSessionLocal()assession:tempselect(Book)filters[]ifsearchInfo.Title:filters.append(Book.Title.like(f%{searchInfo.Title}%))ifsearchInfo.Price:filters.append(Book.PricesearchInfo.Price)iffilters:temptemp.where(and_(*filters))temptemp.order_by(Book.CreateDatetime.desc())temptemp.offset((pageIndex-1)*pageSize).limit(pageSize)resultawaitsession.execute(temp)returnresult.scalars().all() 入口函数 asyncdefmain():# await initDatabase()# 获取所有书籍信息# booksawait queryBooks()print(-------------查询结果--------------)# for book in books:# print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})# 获取第一本书# bookawait queryFirstBook()# print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})# 根据id查询书籍信息# bookawait queryBookById(7805a7af-5a96-4d33-b1a3-9ac9ef48d2a9)# if book:# print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})# else:# print(书籍信息不存在)# 查询价格大于50的书籍# booksawait queryBooksByPrice(50)# for book in books:# print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})# 精准匹配# bookawait queryBookByTitle(红楼梦)# if book:# print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})# else:# print(书籍信息不存在)# 批量筛选书籍# booksawait batchQueryBooks(# [ea1b2895-2210-4666-a4b5-fb68a8ff1c19,# 6b145488-d418-4beb-9ae2-fcfc3ee55391])# for book in books:# print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})# 聚合操作# resultawait jointOperationV2() # jointOperation()# print(结果str(result))# 分页查询searchInfoQueryPagingDatasV3SearchInfo()searchInfo.Price50searchInfo.Title西booksawaitqueryPagingDatasV3(searchInfo,1,3)forbookinbooks:print(f{book.Id}\t{book.Title}\t{book.Author}\t{book.Price}\t{book.CreateDatetime})awaitasyncEngine.dispose()print(OK)if__name____main__:asyncio.run(main())