在 FastAPI 项目开发中，数据库操作是核心环节之一。SQLAlchemy 作为功能强大的 ORM 框架，其异步特性能够很好地适配 FastAPI 的异步架构，提升数据库操作效率。本文将详细讲解如何基于 FastAPI + SQLAlchemy 异步引擎，实现 MySQL 数据库表的自动创建，涵盖依赖安装、异步引擎配置、数据模型定义、监听器自动建表全流程。

1.环境准备

1.1搭建框架

在PyCharm中新建一个fastapi项目（搭建异步 Web 服务的核心框架）。

1.2安装依赖库

要实现 FastAPI 结合 SQLAlchemy 异步操作 MySQL，我们还需要安装必备的第三方库。

在终端执行以下命令：

安装异步 ORM 框架，处理数据库交互；

pip install  async sqlalchemy

安装MySQL 异步驱动，适配 SQLAlchemy 异步引擎。

pip install aiomysql

2.创建异步数据库引擎：

SQLAlchemy 提供了 create_async_engine 方法用于创建异步数据库引擎，是连接 FastAPI 与 MySQL 的核心桥梁。

2.1 核心参数说明

在配置异步引擎时，我们需要注意以下关键参数：

DATABASE_URL：数据库连接字符串，格式为 数据库类型+异步驱动://用户名:密码@主机:端口/数据库名?字符集；

echo：是否打印 SQL 执行日志，开发环境建议设为 True，便于调试；

pool_size：数据库连接池默认大小，控制常驻连接数；

max_overflow：连接池最大溢出连接数，超出 pool_size 的临时连接数上限。

2.2 代码实现

from sqlalchemy.ext.asyncio import create_async_engine

# 配置 MySQL 异步连接字符串
DATABASE_URL = "mysql+aiomysql://root:root@localhost:3306/fastapi_orm?charset=utf8mb4"

# 创建异步数据库引擎

# pool_size：数据库连接池大小
# max_overflow：数据库连接池最大溢出连接数
# echo：是否开启SQL日志
engine = create_async_engine(
    DATABASE_URL,
    echo=True,         
    pool_size=10,       
    max_overflow=20     
)

注意：需确保 MySQL 服务已启动，且 fastapi_orm 数据库已提前创建。

具体实现步骤如下：
1.在终端连接mysql服务器（输入自己mysql数据库的用户名与密码）

mysql -u用户名 -p密码

2.创建数据库fastapi_orm

create database fastapi_orm character set utf8mb4

3.定义数据模型：基类与用户模型

SQLAlchemy 通过 ORM 方式将数据库表映射为 Python 类，我们先定义包含通用字段的基类，再基于基类实现业务模型（如常见的用户表）。

3.1定义基类

创建一个Base基类，继承 DeclarativeBase，我们可以实现对所有表的通用字段的统一管理，比如创建时间和更新时间。

具体实现代码如下：

from datetime import datetime
from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    # create_time Base类的属性名称。
    create_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now,
                                                  insert_default=func.now())

    update_time: Mapped[datetime] = mapped_column(DateTime, default=datetime.now,
                                                  insert_default=func.now(),
                                                  onupdate=func.now())

    )

基类中 create_time 和 update_time 为所有表的通用字段，通过 insert_default 和 onupdate 实现自动赋值，可以减少代码的冗余。

3.2 定义用户表

基于 Base 基类创建用户表 User，映射数据库中的 t_user 表。

具体实现代码如下：

from sqlalchemy import Integer, String, Float

class User(Base):
    # 数据库表名
    __tablename__ = "t_user"

    # 字段定义：Mapped 标注类型，mapped_column 配置字段属性
    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
        autoincrement=True,
        name="user_id",
        comment="用户编号"
    )
    name: Mapped[str] = mapped_column(
        String(20),
        name="user_name",
        nullable=False,
        comment="用户名"
    )
    password: Mapped[str] = mapped_column(
        String(20),
        name="user_password",
        nullable=False,
        comment="密码"
    )
    salary: Mapped[float] = mapped_column(
        Float(6, 2),
        name="user_salary",
        nullable=False,
        comment="工资"
    )
    birthday: Mapped[datetime] = mapped_column(
        DateTime,
        name="user_birthday",
        nullable=False,
        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具体代码实现

from fastapi import FastAPI

# 初始化 FastAPI 应用
app = FastAPI()

# 定义创建表的异步函数
async def create_tables():
    # 开启异步连接上下文
    async with engine.begin() as conn:
        # 同步执行表创建操作（适配 SQLAlchemy 异步逻辑）
        await conn.run_sync(Base.metadata.create_all)

# 注册启动事件：应用启动时自动执行 init 函数
@app.on_event("startup")
async def init():
    await create_tables()

#（可选）
# 注册路由，测试应用是否正常运行
#from orm.urls import orm_router
#app.include_router(orm_router)

4.3定义路由

我们可以在PyCharm中新建一个 urls.py 文件，定义测试路由，验证项目是否成功启动。

具体代码实现：

from fastapi import APIRouter

orm_router = APIRouter()

@orm_router.get("/")
async def index():
    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数据库，查看表结构，确认表已经自动创建。