资讯详情

深入了解 · 专业解读

Python SqlAlchemy对数据库各种操作整理（MySQL为例）

发布时间：2026/7/5 3:28:27
Python SqlAlchemy对数据库各种操作整理（MySQL为例） 提示文章写完后目录可以自动生成如何生成可参考右边的帮助文档文章目录前言一、装包二、数据库直接生成模型类脚本文件RefeshModels.bat二、我的模型类也就是那个models.py三、DbConfig.py四、main.py重点示例代码在这里面前言这段时间研究都在学习python所以对Python SqlAlchemy对数据库各种操作整理MySQL为例大家相互学习也当是作为笔记以后copy来用。一、装包SqlAlchemypip install “sqlalchemy[asyncio]” aiomysqldb frist方式依赖包pip install pymysql sqlacodegen二、数据库直接生成模型类脚本文件RefeshModels.batecho off chcp 65001 nul echo 正在强制刷新数据库模型先删后建... if exist models.py del /f /q models.py sqlacodegen mysqlpymysql://root:xxxlocalhost/db_py_orm_test --outfile models.py if %errorlevel% equ 0 ( echo 模型已强制覆盖更新 ) else ( echo 生成失败请检查数据库连接或参数。 ) pause1、这个脚本文件里面“models.py”这个就是模型类的文件路径了我这里是根目录你可以根据自己需要去改到其他目录2、数据库名字等连接字符串信息根据自己情况改一下就行3、每次数据库更新了运行一下这个脚本就行二、我的模型类也就是那个models.pyfromtypingimportOptionalimportdatetimefromsqlalchemyimportDateTime,Float,Stringfromsqlalchemy.dialects.mysqlimportINTEGERfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):passclassAuthor(Base):__tablename__authorId:Mapped[str]mapped_column(String(50),primary_keyTrue)Name:Mapped[Optional[str]]mapped_column(String(20))Age:Mapped[Optional[int]]mapped_column(INTEGER(11))classBook(Base):__tablename__bookId:Mapped[str]mapped_column(String(50),primary_keyTrue,commentID)CreateDatetime:Mapped[datetime.datetime]mapped_column(DateTime,nullableFalse,commentCreateDate)Title:Mapped[Optional[str]]mapped_column(String(50),commentTitle)Author_:Mapped[Optional[str]]mapped_column(Author,String(20),commentAuthor)Price:Mapped[Optional[float]]mapped_column(Float,commentPrice)三、DbConfig.pyfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSession connectionStringmysqlaiomysql://root:xxxlocalhost:3306/db_py_orm_test?charsetutf8asyncEnginecreate_async_engine(connectionString,echoFalse,pool_size10,max_overflow20)AsyncSessionLocalasync_sessionmaker(bindasyncEngine,class_AsyncSession,expire_on_commitFalse)asyncdefgetDatabase():asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()exceptException:awaitsession.rollback()raisefinally:awaitsession.close()四、main.py重点示例代码在这里面这里需要注意的是那个异步引擎一定要Disposeimportuuidfromsqlalchemyimporttext,select,update,delete,join,insert,existsimportasynciofromsqlalchemy.sql.operatorsimportand_,or_fromsqlalchemy.testingimportin_fromDbConfigimportAsyncSessionLocal,asyncEnginefrommodelsimportBook,Authorfromdatetimeimportdatetime 获取所有书籍信息 返回值是元组列表 asyncdefgetBooks():asyncwithAsyncSessionLocal()assession:return(awaitsession.execute(text(SELECT * FROM book;))).all() 获取第一条有就返回第一条没有就是none 返回的是元组|None asyncdefgetBookFirst():asyncwithAsyncSessionLocal()assession:return(awaitsession.execute(text(SELECT \*\ FROM \ book \ WHERE \ Price200\ ORDER BY \ CreateDatetime DESC;))).first() 返回一条或没有 当存在多条结果抛出异常 asyncdefgetBookByTitle(title):asyncwithAsyncSessionLocal()assession:return(awaitsession.execute(text(SELECT \*\ FROM \ book \ WHERE \ Title:title;), {title:title})).one_or_none() 单一条件的筛选 asyncdefgetBooksV2():asyncwithAsyncSessionLocal()assession:return(awaitsession.execute(select(Book).where(Book.Price50))).scalars().all() 复杂条件筛选 asyncdefgetBooksV3():asyncwith(AsyncSessionLocal()assession):return(awaitsession.execute(select(Book).where(or_(and_(Book.Price50,Book.Author.like(吴%)),Book.Title红楼梦)))).scalars().all() 更新操作V1 asyncdefupdatePrice():asyncwithAsyncSessionLocal()assession:queryResultawaitsession.execute(update(Book).where(Book.Id981dcd3b-cbcc-4663-823b-87bebf76bfbe).values(Price80))awaitsession.commit()returnqueryResult.rowcount0 更新操作V2 asyncdefupdatePriceV2():asyncwithAsyncSessionLocal()assession:book(awaitsession.execute(select(Book).where(Book.Id981dcd3b-cbcc-4663-823b-87bebf76bfbe)))\.scalars()\.one_or_none()ifnotbook:return书籍信息不存在book.Price80awaitsession.commit()returnOK 删除V1 asyncdefdeleteBook():asyncwithAsyncSessionLocal()assession:queryResultawaitsession.execute(delete(Book).where(Book.Id981dcd3b-cbcc-4663-823b-87bebf76bfbe))awaitsession.commit()returnqueryResult.rowcount0 删除V2 asyncdefdeleteBookV2():asyncwithAsyncSessionLocal()assession:book(awaitsession.execute(select(Book).where(Book.Idea1b2895-2210-4666-a4b5-fb68a8ff1c19)))\.scalars()\.one_or_none()ifnotbook:return书籍信息不存在awaitsession.delete(book)awaitsession.commit()returnOK 联表查询 asyncdefgetBooksV4():asyncwithAsyncSessionLocal()assession:return(awaitsession.execute(select(join(Book,Author,Book.Author_Author.Name)).where(Book.Price50))).all() 联表查询只要部分字段 asyncdefgetBooksV5():asyncwithAsyncSessionLocal()assession:return(awaitsession.execute(select(Book.Title,Author.Name,Author.Age).select_from(join(Book,Author,Book.Author_Author.Name)))).all() 批量新增书籍 asyncdefbatchInsertBooks(books):asyncwithAsyncSessionLocal()assession:forbookinbooks:ifnot(awaitsession.execute(select(exists().where(Author.Namebook[Author])))).scalar():return作者信息不存在if(awaitsession.execute(select(exists().where(Book.Titlebook[Title])))).scalar():return书籍信息已存在queryResultawaitsession.execute(insert(Book).values(books))awaitsession.commit()ifqueryResult.rowcount0:returnOKelse:return新增失败 批量删除书籍 asyncdefbatchDeleteBooks(ids):asyncwithAsyncSessionLocal()assession:queryResultawaitsession.execute(delete(Book).where(Book.Id.in_(ids)))awaitsession.commit()returnqueryResult.rowcount0asyncdefmain():print()# books await getBooks()# for book in books:# print(book)# bookawait getBookFirst()# if book:# print(book)# else:# print(Book not found)# bookawait getBookByTitle(西游记)# if book :# print(book)# else:# print(Book not found)# books await getBooksV3()# for book in books:# print(f{book.Id} {book.Title} {book.Author} {book.Price} {book.CreateDatetime})# if await updatePrice():# print(更新成功)# else:# print(更新失败)# if (await updatePriceV2())!OK:# print(更新失败)# else:# print(更新成功)# if await deleteBook():# print(删除成功)# else:# print(删除失败)# if (await deleteBookV2())!OK:# print(删除失败)# else:# print(删除成功)# books await getBooksV5()# for book in books:# print(book)# result await batchInsertBooks([# {# Id: str(uuid.uuid4()),# Title: 红楼梦,# Author: 曹雪芹,# Price: 60,# CreateDatetime: datetime.now()# }, {# Id: str(uuid.uuid4()),# Title: 西游记,# Author: 吴承恩,# Price: 80,# CreateDatetime: datetime.now()# }])# if resultOK:# print(新增成功)# else:# print(result)ifawaitbatchDeleteBooks([57e2cd30-537f-429b-bb35-d270264d48cd,7805a7af-5a96-4d33-b1a3-9ac9ef48d2a9]):print(删除成功)else:print(删除失败)print()awaitasyncEngine.dispose()print(OK)if__name____main__:asyncio.run(main())

相关新闻