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