在 FastAPI 项目开发中数据库操作是核心环节之一。SQLAlchemy 作为功能强大的 ORM 框架其异步特性能够很好地适配 FastAPI 的异步架构提升数据库操作效率。本文将详细讲解如何基于 FastAPI SQLAlchemy 异步引擎实现 MySQL 数据库表的自动创建涵盖依赖安装、异步引擎配置、数据模型定义、监听器自动建表全流程。1.环境准备1.1搭建框架在PyCharm中新建一个fastapi项目搭建异步 Web 服务的核心框架。1.2安装依赖库要实现 FastAPI 结合 SQLAlchemy 异步操作 MySQL我们还需要安装必备的第三方库。在终端执行以下命令安装异步 ORM 框架处理数据库交互pip installasyncsqlalchemy安装MySQL 异步驱动适配 SQLAlchemy 异步引擎。pip install aiomysql2.创建异步数据库引擎SQLAlchemy 提供了 create_async_engine 方法用于创建异步数据库引擎是连接 FastAPI 与 MySQL 的核心桥梁。2.1 核心参数说明在配置异步引擎时我们需要注意以下关键参数DATABASE_URL数据库连接字符串格式为 数据库类型异步驱动://用户名:密码主机:端口/数据库名?字符集echo是否打印 SQL 执行日志开发环境建议设为 True便于调试pool_size数据库连接池默认大小控制常驻连接数max_overflow连接池最大溢出连接数超出 pool_size 的临时连接数上限。2.2 代码实现fromsqlalchemy.ext.asyncioimportcreate_async_engine# 配置 MySQL 异步连接字符串DATABASE_URLmysqlaiomysql://root:rootlocalhost:3306/fastapi_orm?charsetutf8mb4# 创建异步数据库引擎# pool_size数据库连接池大小# max_overflow数据库连接池最大溢出连接数# echo是否开启SQL日志enginecreate_async_engine(DATABASE_URL,echoTrue,pool_size10,max_overflow20)注意需确保 MySQL 服务已启动且 fastapi_orm 数据库已提前创建。具体实现步骤如下1.在终端连接mysql服务器输入自己mysql数据库的用户名与密码mysql-u用户名-p密码2.创建数据库fastapi_ormcreate database fastapi_orm charactersetutf8mb43.定义数据模型基类与用户模型SQLAlchemy 通过 ORM 方式将数据库表映射为 Python 类我们先定义包含通用字段的基类再基于基类实现业务模型如常见的用户表。3.1定义基类创建一个Base基类继承 DeclarativeBase我们可以实现对所有表的通用字段的统一管理比如创建时间和更新时间。具体实现代码如下fromdatetimeimportdatetimefromsqlalchemyimportDateTime,funcfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnclassBase(DeclarativeBase):# create_time Base类的属性名称。create_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,insert_defaultfunc.now())update_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,insert_defaultfunc.now(),onupdatefunc.now()))基类中 create_time 和 update_time 为所有表的通用字段通过 insert_default 和 onupdate 实现自动赋值可以减少代码的冗余。3.2 定义用户表基于 Base 基类创建用户表 User映射数据库中的 t_user 表。具体实现代码如下fromsqlalchemyimportInteger,String,FloatclassUser(Base):# 数据库表名__tablename__t_user# 字段定义Mapped 标注类型mapped_column 配置字段属性id:Mapped[int]mapped_column(Integer,primary_keyTrue,autoincrementTrue,nameuser_id,comment用户编号)name:Mapped[str]mapped_column(String(20),nameuser_name,nullableFalse,comment用户名)password:Mapped[str]mapped_column(String(20),nameuser_password,nullableFalse,comment密码)salary:Mapped[float]mapped_column(Float(6,2),nameuser_salary,nullableFalse,comment工资)birthday:Mapped[datetime]mapped_column(DateTime,nameuser_birthday,nullableFalse,comment生日)每个字段的详细说明primary_key设置主键autoincrement自增属性name数据库字段名与模型属性名不一致时指定nullable是否允许为空comment字段注释提升数据库可读性。4.自动创建数据库表FastAPI 提供了事件监听机制我们可以在项目启动startup时执行指定逻辑利用该监听机制实现数据库表的自动创建。4.1核心思路1.定义 create_tables 函数通过异步引擎的 run_sync 方法调用 Base.metadata.create_all 创建所有基于 Base 的表2.注册 startup 事件当 FastAPI 应用启动时自动执行 create_tables 函数。4.2具体代码实现fromfastapiimportFastAPI# 初始化 FastAPI 应用appFastAPI()# 定义创建表的异步函数asyncdefcreate_tables():# 开启异步连接上下文asyncwithengine.begin()asconn:# 同步执行表创建操作适配 SQLAlchemy 异步逻辑awaitconn.run_sync(Base.metadata.create_all)# 注册启动事件应用启动时自动执行 init 函数app.on_event(startup)asyncdefinit():awaitcreate_tables()#可选# 注册路由测试应用是否正常运行#from orm.urls import orm_router#app.include_router(orm_router)4.3定义路由我们可以在PyCharm中新建一个 urls.py 文件定义测试路由验证项目是否成功启动。在这里插入代码片具体代码实现fromfastapiimportAPIRouter orm_routerAPIRouter()orm_router.get(/)asyncdefindex():return{message:Hello World}5.运行与结果验证5.1设置端口为项目设置端口可以打开浏览器访问文档地址查看是否成功启动也可以通过测试工具如yaak进行测试。这里设置的端口是8888浏览器的访问地址http://127.0.0.1:8888/docs​5.2在PyCharm中连接mysql数据库在PyCharm中连接2.2中创建的数据库输入自己mysql数据库的用户名和密码测试数据库是否成功连接。数据库连接成功后点击Apply→OK完成数据库连接在 Database 面板可查看fastapi_orm数据库。5.3验证表创建结果启动成功后我们可以看到创建 t_user 表的 SQL 语句。表创建成功会看到如下图片的类似信息。也可以登录mysql数据库查看表结构确认表已经自动创建。