有哪些比較好的做ppt好的網(wǎng)站,韶關(guān)市住房和城鄉(xiāng)建設(shè)管理局網(wǎng)站,微信網(wǎng)站制作系統(tǒng),收錄優(yōu)美圖片崩了原文#xff1a;towardsdatascience.com/how-to-use-sqlalchemy-to-make-database-requests-asynchronously-e90a4c8c11b1 數(shù)據(jù)庫請(qǐng)求是一個(gè)典型的 I/O 密集型任務(wù)#xff0c;因?yàn)樗蟛糠謺r(shí)間都在等待數(shù)據(jù)庫服務(wù)器的響應(yīng)。因此#xff0c;如果你的應(yīng)用程序進(jìn)行了大量的數(shù)據(jù)…原文towardsdatascience.com/how-to-use-sqlalchemy-to-make-database-requests-asynchronously-e90a4c8c11b1數(shù)據(jù)庫請(qǐng)求是一個(gè)典型的 I/O 密集型任務(wù)因?yàn)樗蟛糠謺r(shí)間都在等待數(shù)據(jù)庫服務(wù)器的響應(yīng)。因此如果你的應(yīng)用程序進(jìn)行了大量的數(shù)據(jù)庫請(qǐng)求那么通過并發(fā)執(zhí)行它們性能可以得到顯著提升這是 SQLAlchemy一個(gè)多功能的 Python SQL 工具包和對(duì)象關(guān)系映射器所支持的。此外異步編程在 Python 中越來越受歡迎尤其是在使用 FastAPI 進(jìn)行 Web 開發(fā)時(shí)我們經(jīng)常需要在協(xié)程中執(zhí)行數(shù)據(jù)庫請(qǐng)求即在用async def語句定義的函數(shù)中。不幸的是我們不能使用經(jīng)典的同步版本的 SQLAlchemy而需要?jiǎng)?chuàng)建引擎、連接和會(huì)話的異步版本。在本文中我們將介紹如何在不同的場(chǎng)景下使用 SQLAlchemy 異步即使用簡(jiǎn)單的 SQL 查詢、Core 和 ORM。重要的是我們將介紹如何在多個(gè)異步任務(wù)中并發(fā)使用它如果使用得當(dāng)可以顯著提高 I/O 密集型應(yīng)用程序的效率。準(zhǔn)備工作我們將使用 Docker 在本地啟動(dòng)一個(gè) MySQL 服務(wù)器在其中我們將創(chuàng)建用于演示的數(shù)據(jù)庫和表# Create a volume to persist the data.$ docker volume create mysql8-data# Create the container for MySQL.$ docker run--name mysql8-d-e MYSQL_ROOT_PASSWORDroot-p13306:3306-v mysql8-data:/var/lib/mysql mysql:8# Connect to the local MySQL server in Docker.$ dockerexec-it mysql8 mysql-u root-proot mysqlSELECT VERSION();-----------|VERSION()|-----------|8.3.0|-----------1rowinset(0.00sec)CREATE DATABASE sales;CREATE TABLE sales.customers(id SMALLINT NOT NULL AUTO_INCREMENT,name VARCHAR(50)NOT NULL,job VARCHAR(50)DEFAULT,PRIMARY KEY(id),UNIQUE UQ_name(name));INSERT INTO sales.customers(name,job)VALUES(Lynn,Backend Developer);然后讓我們創(chuàng)建一個(gè) 虛擬環(huán)境這樣我們就可以嘗試最新的 Python 和庫版本conda create-n sql python3.12conda activate sql pip install-Usqlalchemy[asyncio]2.0,2.1pip install-Uaiomysql0.2,0.3pip install-Ucryptography42.0,42.1sqlalchemy[asyncio]– SQLAlchemy 與greenlet依賴項(xiàng)一起安裝這是一個(gè) SQLAlchemy 用于異步工作的庫。aiomysql– 一個(gè)從 asyncio 框架訪問 MySQL 數(shù)據(jù)庫的驅(qū)動(dòng)程序它背后使用 PyMySQL。cryptography– 由 SQLAlchemy 用于身份驗(yàn)證。異步執(zhí)行簡(jiǎn)單的 SQL 查詢要使用 SQLAlchemy 異步運(yùn)行 SQL 查詢我們首先需要使用create_async_engine()創(chuàng)建一個(gè)異步引擎。然后在創(chuàng)建連接、執(zhí)行查詢和處置引擎時(shí)我們需要使用awaitimportasynciofromsqlalchemyimporttextfromsqlalchemy.ext.asyncioimportcreate_async_engineasyncdefmain():# Create an asynchronous engine.async_enginecreate_async_engine(mysqlaiomysql://root:rootlocalhost:13306/sales)# Insert new data with a transation.asyncwithasync_engine.begin()asconn:insert_querytext( INSERT INTO sales.customers (name, job) VALUES (:name, :job) )awaitconn.execute(insert_query,{name:Hans,job:Data Engineer})# Check the data afer its inserted.asyncwithasync_engine.connect()asconn:select_querytext( SELECT * FROM sales.customers WHERE name :name )resultawaitconn.execute(select_query,{name:Hans})print(result.fetchall())# Close and clean-up pooled connections.awaitasync_engine.dispose()asyncio.run(main())注意當(dāng)異步執(zhí)行如上所示的簡(jiǎn)單 SQL 查詢時(shí)我們需要使用字典傳遞變量而不是像同步版本那樣使用關(guān)鍵字參數(shù)。當(dāng)運(yùn)行上面的代碼時(shí)你將看到以下結(jié)果打印出來[(2,Hans,Data Engineer))]當(dāng)你想快速開始使用 SQLAlchemy 而又不了解 Core 和 ORM 功能時(shí)使用簡(jiǎn)單的 SQL 查詢是一個(gè)不錯(cuò)的選擇。然而如你所見它并不太符合 Python 風(fēng)格因?yàn)樗褂昧俗杂娠L(fēng)格的簡(jiǎn)單 SQL 查詢。當(dāng)你對(duì) SQLAlchemy 有更多經(jīng)驗(yàn)時(shí)你可能想使用 Core 或 ORM 功能。使用 SQLAlchemy Core 異步執(zhí)行在 SQLAlchemy 2.0 中核心功能通常意味著直接與Table對(duì)象交互現(xiàn)在非常強(qiáng)大。它實(shí)際上與 ORM 功能混合到了非常高的程度。例如select操作符可以用于核心和 ORM。對(duì)于核心使用我們還需要?jiǎng)?chuàng)建一個(gè)異步引擎然后使用它來異步創(chuàng)建連接?；竟ぷ髁鞒膛c普通查詢相同不同之處在于語句是通過核心操作符如insert和select構(gòu)建的。importasynciofromsqlalchemyimportColumn,Integer,insertfromsqlalchemyimportMetaDatafromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemyimportTablefromsqlalchemy.ext.asyncioimportcreate_async_engine meta_dataMetaData()tableTable(customers,meta_data,Column(id,Integer,primary_keyTrue),Column(name,String(50),nullableFalse),Column(job,String(50),default),)asyncdefmain():# Create an asynchronous engine.async_enginecreate_async_engine(mysqlaiomysql://root:rootlocalhost:13306/sales)# Insert new data with a transation.asyncwithengine.begin()asconn:stmtinsert(table).values(nameJack,jobFrontend Developer)awaitconn.execute(stmt)# Check the data afer its inserted.asyncwithengine.connect()asconn:resultawaitconn.execute(select(table).where(table.c.nameJack))print(result.fetchall())# Close and clean-up pooled connections.awaitengine.dispose()asyncio.run(main())當(dāng)運(yùn)行上述代碼時(shí)將顯示以下結(jié)果[(3,Jack,Frontend Developer)]使用 SQLAlchemy ORM 異步使用 SQLAlchemy ORM 的 ORM 功能要復(fù)雜一些尤其是在 2.0 版本中ORM 類的創(chuàng)建語法發(fā)生了顯著變化。特別是Mapped[]用于指定類型mapped_column()構(gòu)造其他列屬性。fromsqlalchemyimportStringfromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportmapped_columnclassBase(DeclarativeBase):passclassCustomer(Base):__tablename__customersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse,uniqueTrue)job:Mapped[str|None]mapped_column(String(50),default)要異步處理 ORM我們需要使用async_sessionmaker()創(chuàng)建一個(gè)異步會(huì)話工廠然后使用with來創(chuàng)建異步會(huì)話實(shí)例# Create an asynchronous session.async_sessionasync_sessionmaker(engine,expire_on_commitFalse)# Create an async session instance.asyncwithasync_session()assession:...處理 ORM 的完整異步代碼如下importasynciofromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemy.ext.asyncioimportasync_sessionmakerfromsqlalchemy.ext.asyncioimportcreate_async_enginefromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportmapped_columnclassBase(DeclarativeBase):passclassCustomer(Base):__tablename__customersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse,uniqueTrue)job:Mapped[str|None]mapped_column(String(50),default)asyncdefmain():# Create an asynchronous engine.enginecreate_async_engine(mysqlaiomysql://root:rootlocalhost:13306/sales)# Create an asynchronous session.async_sessionasync_sessionmaker(engine,expire_on_commitFalse)# Create an async session instance.asyncwithasync_session()assession:# Insert new data with a transation.asyncwithsession.begin():session.add(Customer(nameStephen,jobManager))# Check the data afer its inserted.asyncwithasync_session()assession:resultawaitsession.execute(select(Customer).where(Customer.nameStephen))customerresult.scalars().one()print(fname {customer.name}, job {customer.job})# Close and clean-up pooled connections.awaitengine.dispose()asyncio.run(main())當(dāng)運(yùn)行上述代碼時(shí)將顯示以下結(jié)果nameStephen,jobManager使用 SQLAlchemy Core 在多個(gè)異步任務(wù)中在多個(gè)異步任務(wù)中并發(fā)使用 SQLAlchemy Core 簡(jiǎn)單因?yàn)檫B接對(duì)象可以直接在多個(gè)異步任務(wù)中傳遞和使用importasynciofrompprintimportpprintfromsqlalchemyimportColumn,IntegerfromsqlalchemyimportMetaDatafromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemyimportTablefromsqlalchemy.ext.asyncioimportcreate_async_engine meta_dataMetaData()tableTable(customers,meta_data,Column(id,Integer,primary_keyTrue),Column(name,String(50),nullableFalse),Column(job,String(50),default),)asyncdefget_customer(name,conn):resultawaitconn.execute(select(table).where(table.c.namename))returnresult.fetchone()asyncdefmain():# Create an asynchronous engine.enginecreate_async_engine(mysqlaiomysql://root:rootlocalhost:13306/sales)names[Lynn,Hans,Jack,Stephen]tasks[]# Check the data afer its inserted.asyncwithengine.connect()asconn:fornameinnames:tasks.append(get_customer(name,conn))resultsawaitasyncio.gather(*tasks)pprint(results)# Close and clean-up pooled connections.awaitengine.dispose()asyncio.run(main())當(dāng)運(yùn)行上述代碼時(shí)你會(huì)看到以下結(jié)果打印出來[(1,Lynn,Backend Developer),(2,Hans,Data Engineer),(3,Jack,Frontend Developer),(4,Stephen,Manager)]在多個(gè)異步任務(wù)中使用 SQLAlchemy ORM另一方面在多個(gè)異步任務(wù)中使用 SQLAlchemy ORM 要復(fù)雜一些因?yàn)椴荒苤苯釉诓l(fā)任務(wù)中使用相同的AsyncSession實(shí)例。讓我們直接嘗試使用它并看看會(huì)發(fā)生什么importasynciofromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemy.ext.asyncioimportasync_sessionmakerfromsqlalchemy.ext.asyncioimportcreate_async_enginefromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportmapped_columnclassBase(DeclarativeBase):passclassCustomer(Base):__tablename__customersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse,uniqueTrue)job:Mapped[str|None]mapped_column(String(50),default)asyncdefget_customer(name,session):resultawaitsession.execute(select(Customer).where(Customer.namename))customerresult.scalars().one()return{name:customer.name,job:customer.job}asyncdefmain():# Create an asynchronous engine.enginecreate_async_engine(mysqlaiomysql://root:rootlocalhost:13306/sales)# Create an asynchronous session.async_sessionasync_sessionmaker(engine,expire_on_commitFalse)names[Lynn,Hans,Jack,Stephen]tasks[]# Check the data afer its inserted.asyncwithasync_session()assession:fornameinnames:tasks.append(get_customer(name,session))resultsawaitasyncio.gather(*tasks)print(results)# Close and clean-up pooled connections.awaitengine.dispose()asyncio.run(main())當(dāng)運(yùn)行上述代碼時(shí)你會(huì)看到以下錯(cuò)誤sqlalchemy.exc.InvalidRequestError:This sessionisprovisioning a new connection;concurrent operations arenotpermitted這個(gè)錯(cuò)誤意味著單個(gè)AsyncSession實(shí)例不能在多個(gè)并發(fā)任務(wù)例如使用asyncio.gather()之類的函數(shù)之間共享。如果你想深入了解這個(gè)話題可以查看這個(gè)參考。解決這個(gè)問題的簡(jiǎn)單可行方案是在每個(gè)任務(wù)中創(chuàng)建一個(gè)AsyncSession實(shí)例。我們將重構(gòu)代碼以全局創(chuàng)建engine和async_session_factory然后在每個(gè)任務(wù)中調(diào)用async_session_factory()來創(chuàng)建一個(gè)獨(dú)立的會(huì)話importasynciofrompprintimportpprintfromsqlalchemyimportselectfromsqlalchemyimportStringfromsqlalchemy.ext.asyncioimportasync_sessionmakerfromsqlalchemy.ext.asyncioimportcreate_async_enginefromsqlalchemy.ormimportDeclarativeBasefromsqlalchemy.ormimportMappedfromsqlalchemy.ormimportmapped_column# Create an asynchronous engine.enginecreate_async_engine(mysqlaiomysql://root:rootlocalhost:13306/sales)# Create an asynchronous session.async_session_factoryasync_sessionmaker(engine,expire_on_commitFalse)classBase(DeclarativeBase):passclassCustomer(Base):__tablename__customersid:Mapped[int]mapped_column(primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse,uniqueTrue)job:Mapped[str|None]mapped_column(String(50),default)asyncdefget_customer(name):# Create an async session instance.asyncwithasync_session_factory()assession:resultawaitsession.execute(select(Customer).where(Customer.namename))customerresult.scalars().one()return{name:customer.name,job:customer.job}asyncdefmain():names[Lynn,Hans,Jack,Stephen]tasks[]# Check the data afer its inserted.fornameinnames:tasks.append(get_customer(name))resultsawaitasyncio.gather(*tasks)pprint(results)# Close and clean-up pooled connections.awaitengine.dispose()asyncio.run(main())當(dāng)代碼運(yùn)行時(shí)你會(huì)看到以下結(jié)果打印出來[{job:Backend Developer,name:Lynn},{job:Data Engineer,name:Hans},{job:Frontend Developer,name:Jack},{job:Manager,name:Stephen}]就像 HTTP 請(qǐng)求一樣數(shù)據(jù)庫請(qǐng)求也是 I/O 密集型任務(wù)因?yàn)樗鼈兇蟛糠謺r(shí)間都在等待數(shù)據(jù)庫服務(wù)器的響應(yīng)。因此我們可以通過并發(fā)而不是順序地執(zhí)行數(shù)據(jù)庫請(qǐng)求來顯著提高應(yīng)用程序的效率。另一方面異步地執(zhí)行數(shù)據(jù)庫請(qǐng)求也越來越重要因?yàn)楫惒骄幊淘?Python 中變得越來越流行尤其是在使用 FastAPI 進(jìn)行 Web 開發(fā)時(shí)這也突出了學(xué)習(xí)這個(gè)主題的必要性。在這篇文章中我們介紹了如何在不同的場(chǎng)景下使用 SQLAlchemy 進(jìn)行異步操作即使用純 SQL 查詢、Core 和 ORM。你可以簡(jiǎn)單地調(diào)整代碼以適應(yīng)你的特定使用。我們特別介紹了如何在多個(gè)異步任務(wù)中并發(fā)使用 SQLAlchemy如果應(yīng)用程序需要并發(fā)執(zhí)行大量數(shù)據(jù)庫請(qǐng)求這將提高應(yīng)用程序的效率。相關(guān)文章學(xué)習(xí)基礎(chǔ)知識(shí)并開始使用 SQLAlchemy ORM如何在 Python 中使用 SQLAlchemy 執(zhí)行純 SQL 查詢