FastAPI 集成 SQLAlchemy2.0 異步數(shù)據(jù)庫(kù)主流 Python ORM 對(duì)比Python 生態(tài)有很多 ORM 庫(kù)SQLAlchemy、Peewee、Pony ORM、Tortoise?ORM。SQLAlchemy生態(tài)最成熟支持異步社區(qū)活躍SQLModelFastAPI 作者開(kāi)發(fā)底層封裝 SQLAlchemy Core目前還不夠成熟ORM 性能測(cè)試參考https://github.com/tortoise/orm?benchmarks/tree/master/master一、安裝異步 SQLAlchemypip install sqlalchemy[asyncio]2.0.31二、創(chuàng)建數(shù)據(jù)庫(kù)連接1. 連接字符串格式mysqlasyncmy://用戶名:密碼主機(jī):端口/數(shù)據(jù)庫(kù)名?charsetutf8mb4示例DB_URI mysqlasyncmy://root:root127.0.0.1:3306/book_db?charsetutf8mb42. 創(chuàng)建異步 EngineEngine 負(fù)責(zé)管理連接池、SQL 翻譯全局只實(shí)例化一個(gè)對(duì)象。from sqlalchemy.ext.asyncio import create_async_engine engine create_async_engine( mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4, echoTrue, # 打印執(zhí)行SQL日志生產(chǎn)環(huán)境關(guān)閉 pool_size10, # 連接池常駐連接數(shù)量默認(rèn)5 max_overflow20, # 連接池最大允許額外創(chuàng)建連接數(shù)默認(rèn)10 pool_timeout10, # 獲取連接超時(shí)時(shí)間單位秒 pool_recycle3600, # 連接回收時(shí)間避免MySQL斷開(kāi)空閑連接 pool_pre_pingTrue # 獲取連接前探測(cè)連接是否有效 )參數(shù)說(shuō)明echoTrue開(kāi)發(fā)調(diào)試打印全部 SQL線上務(wù)必關(guān)閉。pool_pre_pingTrue自動(dòng)檢測(cè)斷開(kāi)的數(shù)據(jù)庫(kù)連接避免報(bào)連接失效錯(cuò)誤。3. 創(chuàng)建異步會(huì)話工廠 Session使用sessionmaker生成會(huì)話工廠后續(xù)每次數(shù)據(jù)庫(kù)操作都從工廠拿AsyncSession會(huì)話對(duì)象。from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.asyncio import AsyncSession AsyncSessionFactory sessionmaker( bindengine, class_AsyncSession, # 使用異步Session autoflushTrue, expire_on_commitFalse )expire_on_commitFalsecommit 之后對(duì)象屬性不會(huì)過(guò)期FastAPI 開(kāi)發(fā)常用配置。接下來(lái)就可以基于會(huì)話工廠去定義 ORM 模型實(shí)現(xiàn)數(shù)據(jù)庫(kù) CRUD。小結(jié)FastAPI 異步 MySQL驅(qū)動(dòng)使用asyncmy安裝sqlalchemy[asyncio]。create_async_engine創(chuàng)建全局唯一異步引擎配置連接池參數(shù)優(yōu)化性能。sessionmaker指定class_AsyncSession得到異步會(huì)話工廠后續(xù)操作數(shù)據(jù)庫(kù)都使用會(huì)話。