FastAPI 集成 SQLAlchemy2.0 异步数据库主流 Python ORM 对比Python 生态有很多 ORM 库SQLAlchemy、Peewee、Pony ORM、Tortoise‑ORM。SQLAlchemy生态最成熟支持异步社区活跃SQLModelFastAPI 作者开发底层封装 SQLAlchemy Core目前还不够成熟ORM 性能测试参考https://github.com/tortoise/orm‑benchmarks/tree/master/master一、安装异步 SQLAlchemypip install sqlalchemy[asyncio]2.0.31二、创建数据库连接1. 连接字符串格式mysqlasyncmy://用户名:密码主机:端口/数据库名?charsetutf8mb4示例DB_URI mysqlasyncmy://root:root127.0.0.1:3306/book_db?charsetutf8mb42. 创建异步 EngineEngine 负责管理连接池、SQL 翻译全局只实例化一个对象。from sqlalchemy.ext.asyncio import create_async_engine engine create_async_engine( mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4, echoTrue, # 打印执行SQL日志生产环境关闭 pool_size10, # 连接池常驻连接数量默认5 max_overflow20, # 连接池最大允许额外创建连接数默认10 pool_timeout10, # 获取连接超时时间单位秒 pool_recycle3600, # 连接回收时间避免MySQL断开空闲连接 pool_pre_pingTrue # 获取连接前探测连接是否有效 )参数说明echoTrue开发调试打印全部 SQL线上务必关闭。pool_pre_pingTrue自动检测断开的数据库连接避免报连接失效错误。3. 创建异步会话工厂 Session使用sessionmaker生成会话工厂后续每次数据库操作都从工厂拿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 开发常用配置。接下来就可以基于会话工厂去定义 ORM 模型实现数据库 CRUD。小结FastAPI 异步 MySQL驱动使用asyncmy安装sqlalchemy[asyncio]。create_async_engine创建全局唯一异步引擎配置连接池参数优化性能。sessionmaker指定class_AsyncSession得到异步会话工厂后续操作数据库都使用会话。
FastAPI 集成 SQLAlchemy2.0 异步数据库主流 Python ORM 对比Python 生态有很多 ORM 库SQLAlchemy、Peewee、Pony ORM、Tortoise‑ORM。SQLAlchemy生态最成熟支持异步社区活跃SQLModelFastAPI 作者开发底层封装 SQLAlchemy Core目前还不够成熟ORM 性能测试参考https://github.com/tortoise/orm‑benchmarks/tree/master/master一、安装异步 SQLAlchemypip install sqlalchemy[asyncio]2.0.31二、创建数据库连接1. 连接字符串格式mysqlasyncmy://用户名:密码主机:端口/数据库名?charsetutf8mb4示例DB_URI mysqlasyncmy://root:root127.0.0.1:3306/book_db?charsetutf8mb42. 创建异步 EngineEngine 负责管理连接池、SQL 翻译全局只实例化一个对象。from sqlalchemy.ext.asyncio import create_async_engine engine create_async_engine( mysqlasyncmy://root:root127.0.0.1:3306/sqlalchemy_learn?charsetutf8mb4, echoTrue, # 打印执行SQL日志生产环境关闭 pool_size10, # 连接池常驻连接数量默认5 max_overflow20, # 连接池最大允许额外创建连接数默认10 pool_timeout10, # 获取连接超时时间单位秒 pool_recycle3600, # 连接回收时间避免MySQL断开空闲连接 pool_pre_pingTrue # 获取连接前探测连接是否有效 )参数说明echoTrue开发调试打印全部 SQL线上务必关闭。pool_pre_pingTrue自动检测断开的数据库连接避免报连接失效错误。3. 创建异步会话工厂 Session使用sessionmaker生成会话工厂后续每次数据库操作都从工厂拿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 开发常用配置。接下来就可以基于会话工厂去定义 ORM 模型实现数据库 CRUD。小结FastAPI 异步 MySQL驱动使用asyncmy安装sqlalchemy[asyncio]。create_async_engine创建全局唯一异步引擎配置连接池参数优化性能。sessionmaker指定class_AsyncSession得到异步会话工厂后续操作数据库都使用会话。
