FastAPI + MySQL 实战：从零搭建带数据库的 CRUD 接口

在上一篇 FastAPI 入门教程中，我们已经学会了如何创建 Web API、定义路由和处理请求参数。不过，如果数据只保存在 Python 的内存字典中，应用重启后数据就会丢失，无法满足真实项目的需求。

因此，本文将进一步介绍如何使用 FastAPI 连接 MySQL 数据库，通过 SQLAlchemy 2.0 实现完整的 CRUD 操作，并使用 Pydantic 进行数据校验。

完成本教程后，你将拥有一个可以创建、查询、修改和删除任务的 Web API，并且数据会持久保存在 MySQL 中。

一、项目技术栈

本项目采用以下技术：

SQLAlchemy ORM 可以让开发者通过 Python 类和对象操作数据库，减少直接编写 SQL 的工作量，同时保留必要的 SQL 查询能力。

本教程使用同步数据库连接，便于初学者理解。对于一般规模的入门项目，这是一种清晰、实用的起步方式。

二、创建项目并安装依赖

1. 创建项目目录

在终端中执行：

mkdir fastapi-mysql-demo
cd fastapi-mysql-demo

创建虚拟环境。

Windows：

python -m venv .venv
.venv\Scripts\activate

Debian 或其他 Linux 系统：

python3 -m venv .venv
source .venv/bin/activate

如果 Debian 缺少虚拟环境组件，可以执行：

sudo apt update
sudo apt install python3-venv

2. 安装项目依赖

python -m pip install "fastapi[standard]" sqlalchemy pymysql

如果你的环境使用 python3 命令，也可以相应地使用：

python3 -m pip install "fastapi[standard]" sqlalchemy pymysql

安装完成后，可以保存依赖清单：

python -m pip freeze > requirements.txt

这样，在其他环境部署时，就可以根据依赖文件重新安装项目所需的软件包。

三、配置 MySQL 数据库

1. 创建数据库

首先确保 MySQL 已经安装并运行。

使用具有创建数据库权限的 MySQL 账户登录：

mysql -u root -p

输入密码后，执行以下 SQL：

CREATE DATABASE fastapi_demo
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;

这里创建了名为 fastapi_demo 的数据库，并使用 utf8mb4 字符集，以便正确存储中文和其他 Unicode 字符。

2. 创建专用数据库账户

正式项目不建议直接使用 MySQL 的 root 账户连接应用。可以创建一个专门用于本项目的数据库用户：

CREATE USER 'fastapi_user'@'localhost'
IDENTIFIED BY '替换为高强度随机密码';

GRANT SELECT, INSERT, UPDATE, DELETE
ON fastapi_demo.*
TO 'fastapi_user'@'localhost';

FLUSH PRIVILEGES;

请将示例密码替换为真实的强密码。

上述授权适用于应用连接本机 MySQL 的常见场景。如果应用需要执行数据库迁移，还应根据迁移方式单独规划所需的 DDL 权限，而不是直接授予所有权限。

如果 FastAPI 和 MySQL 不在同一台机器上，数据库账户的主机匹配规则也需要根据实际网络环境调整。

3. 测试数据库连接

退出 MySQL 后，可以使用新账户测试：

mysql -u fastapi_user -p fastapi_demo

如果能够成功进入数据库，说明数据库及账户配置基本完成。

四、设计项目目录结构

为了避免所有代码都堆积在一个文件中，建议采用以下结构：

fastapi-mysql-demo/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   └── schemas.py
├── .env
├── .gitignore
└── requirements.txt

各文件的作用如下：

• main.py：定义 API 路由和业务处理逻辑。

• database.py：管理数据库连接、会话和依赖注入。

• models.py：定义 SQLAlchemy ORM 数据模型。

• schemas.py：定义请求体和响应数据结构。

• .env：保存本地环境配置。

• .gitignore：防止提交虚拟环境和敏感配置。

这种结构适合小型示例项目。后续增加用户管理、文章管理等模块时，可以继续拆分 routers、services 和 repositories。

五、配置数据库连接

在项目根目录创建 .env 文件：

DB_HOST=127.0.0.1
DB_PORT=3306
DB_NAME=fastapi_demo
DB_USER=fastapi_user
DB_PASSWORD=替换为高强度随机密码

这里的配置值需要根据你的 MySQL 实际环境调整。

接下来修改 app/database.py：

import os
from urllib.parse import quote_plus

from sqlalchemy import URL, create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker


def get_database_url():
    password = os.environ["DB_PASSWORD"]

    return URL.create(
        drivername="mysql+pymysql",
        username=os.environ["DB_USER"],
        password=password,
        host=os.environ.get("DB_HOST", "127.0.0.1"),
        port=int(os.environ.get("DB_PORT", "3306")),
        database=os.environ["DB_NAME"],
        query={"charset": "utf8mb4"},
    )


class Base(DeclarativeBase):
    pass


engine = create_engine(
    get_database_url(),
    pool_pre_ping=True,
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
)


def get_db():
    db = SessionLocal()

    try:
        yield db
    finally:
        db.close()

注意：SQLAlchemy 的 URL.create() 会处理数据库连接 URL 中的特殊字符，因此不需要手动拼接密码字符串。上面导入的 quote_plus 并未使用，可以删除该导入，不影响程序运行。

本示例使用操作系统环境变量读取配置。.env 文件不会被 Python 自动读取，因此还需要在启动前加载环境变量。

在开发环境加载 .env

安装 python-dotenv：

python -m pip install python-dotenv

在 app/database.py 顶部增加：

from dotenv import load_dotenv

load_dotenv()

这样，应用启动时就会尝试从项目根目录的 .env 文件读取配置。

正式部署时，也可以直接由 systemd、容器平台或其他进程管理工具注入环境变量。请勿将包含真实数据库密码的 .env 文件提交到公开代码仓库。

六、创建 SQLAlchemy 数据模型

接下来定义任务表。

在 app/models.py 中编写：

from datetime import datetime

from sqlalchemy import Boolean, DateTime, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(
        primary_key=True,
        index=True,
    )

    title: Mapped[str] = mapped_column(
        String(100),
        nullable=False,
    )

    description: Mapped[str | None] = mapped_column(
        Text,
        nullable=True,
    )

    completed: Mapped[bool] = mapped_column(
        Boolean,
        default=False,
        nullable=False,
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        server_default=func.now(),
        nullable=False,
    )

这里定义了一个名为 Task 的 ORM 模型，对应数据库中的 tasks 表。

字段说明：

SQLAlchemy 2.0 使用 Mapped 和 mapped_column() 定义类型明确的 ORM 字段。

需要注意，default=False 是 SQLAlchemy 提供的 Python 端默认值，而 server_default 则是在数据库服务器端设置默认值。实际项目中应根据数据写入方式明确默认值由哪一层负责。

七、创建 Pydantic 请求与响应模型

数据库模型和 API 数据模型承担不同职责。

数据库模型描述表结构，而 Pydantic 模型负责校验客户端输入、定义 API 输出格式。

在 app/schemas.py 中编写：

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=100)
    description: str | None = None
    completed: bool = False


class TaskUpdate(BaseModel):
    title: str | None = Field(
        default=None,
        min_length=1,
        max_length=100,
    )
    description: str | None = None
    completed: bool | None = None


class TaskResponse(BaseModel):
    id: int
    title: str
    description: str | None
    completed: bool
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

三个模型的区别：

• TaskCreate：创建任务时使用，要求提供有效标题。

• TaskUpdate：修改任务时使用，允许只更新部分字段。

• TaskResponse：向客户端返回任务信息。

from_attributes=True 允许 Pydantic 从 SQLAlchemy ORM 对象的属性读取数据并构建响应模型。

本例允许 description 为 None。如果需要限制描述长度，可以在创建和更新模型中为它增加相应的 Field 约束。

八、实现完整 CRUD 接口

现在将数据库模型、连接会话和 Pydantic 模型组合起来。

在 app/main.py 中编写：

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from fastapi import Depends, FastAPI, HTTPException, Response, status
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.database import Base, engine, get_db
from app.models import Task
from app.schemas import TaskCreate, TaskResponse, TaskUpdate


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # 仅用于本教程的入门演示。
    # 正式项目建议使用 Alembic 管理数据库迁移。
    Base.metadata.create_all(bind=engine)
    yield


app = FastAPI(
    title="FastAPI + MySQL 任务管理 API",
    version="1.0.0",
    lifespan=lifespan,
)


@app.get("/")
def home():
    return {"message": "FastAPI + MySQL 运行正常"}


@app.post(
    "/tasks",
    response_model=TaskResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_task(
    payload: TaskCreate,
    db: Session = Depends(get_db),
):
    task = Task(**payload.model_dump())

    db.add(task)
    db.commit()
    db.refresh(task)

    return task


@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(
    skip: int = 0,
    limit: int = 20,
    db: Session = Depends(get_db),
):
    if skip < 0:
        raise HTTPException(
            status_code=400,
            detail="skip 不能小于 0",
        )

    if not 1 <= limit <= 100:
        raise HTTPException(
            status_code=400,
            detail="limit 必须在 1 到 100 之间",
        )

    statement = (
        select(Task)
        .order_by(Task.id.desc())
        .offset(skip)
        .limit(limit)
    )

    return db.scalars(statement).all()


@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(
    task_id: int,
    db: Session = Depends(get_db),
):
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(
            status_code=404,
            detail="任务不存在",
        )

    return task


@app.patch("/tasks/{task_id}", response_model=TaskResponse)
def update_task(
    task_id: int,
    payload: TaskUpdate,
    db: Session = Depends(get_db),
):
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(
            status_code=404,
            detail="任务不存在",
        )

    changes = payload.model_dump(exclude_unset=True)

    for field, value in changes.items():
        setattr(task, field, value)

    db.commit()
    db.refresh(task)

    return task


@app.delete(
    "/tasks/{task_id}",
    status_code=status.HTTP_204_NO_CONTENT,
)
def delete_task(
    task_id: int,
    db: Session = Depends(get_db),
):
    task = db.get(Task, task_id)

    if task is None:
        raise HTTPException(
            status_code=404,
            detail="任务不存在",
        )

    db.delete(task)
    db.commit()

    return Response(status_code=status.HTTP_204_NO_CONTENT)

代码中的关键点

1. 使用 Depends(get_db) 获取数据库会话

FastAPI 会调用 get_db()，在请求处理过程中提供数据库会话，并在依赖清理时关闭会话。

2. 创建记录后执行 commit()

db.add(task) 将对象加入当前会话，db.commit() 提交事务，db.refresh(task) 则重新读取数据库生成或更新的字段。

3. 使用 db.get() 查询单条记录

db.get(Task, task_id) 根据主键查找任务。如果没有对应记录，就返回 None。

4. 使用 exclude_unset=True 实现部分更新

例如，客户端只提交：

{
  "completed": true
}

程序只会修改 completed 字段，不会因为其他字段未提交而将其覆盖为默认值。

需要注意，显式提交 "description": null 与完全不提交 description 是不同的操作：前者会把描述设置为空值，后者则不会修改描述。

5. 删除操作返回 HTTP 204

删除成功后，接口返回 204 No Content，表示操作成功且没有响应正文。

九、启动项目并测试接口

在项目根目录执行：

fastapi dev app/main.py

如果数据库配置正确、MySQL 正常运行，应用启动时会尝试创建示例所需的 tasks 表。

然后访问：

http://127.0.0.1:8000/docs

FastAPI 会自动生成 Swagger UI 文档，可以在浏览器中直接测试全部接口。

1. 创建任务

请求：

POST /tasks
Content-Type: application/json

请求体：

{
  "title": "学习 FastAPI",
  "description": "完成数据库 CRUD 教程",
  "completed": false
}

预期返回类似：

{
  "id": 1,
  "title": "学习 FastAPI",
  "description": "完成数据库 CRUD 教程",
  "completed": false,
  "created_at": "2026-10-30T10:00:00"
}

实际 ID 和时间由数据库决定。

2. 查询任务列表

请求：

GET /tasks?skip=0&limit=20

返回任务数组，支持通过 skip 和 limit 控制分页。

3. 查询单个任务

请求：

GET /tasks/1

如果 ID 为 1 的任务存在，就返回该任务；如果不存在，则返回 HTTP 404。

4. 修改任务

请求：

PATCH /tasks/1
Content-Type: application/json

请求体：

{
  "completed": true
}

返回更新后的任务对象。

5. 删除任务

请求：

DELETE /tasks/1

如果任务存在，接口返回 HTTP 204；如果任务不存在，则返回 HTTP 404。

十、常见错误及解决方法

1. Access denied for user

常见原因包括数据库用户名或密码错误、账户主机匹配规则不正确，以及账户缺少所需权限。

建议使用项目专用数据库账户重新测试连接，并检查 MySQL 的账户授权。

2. Can't connect to MySQL server

可能是 MySQL 服务没有启动、端口配置错误，或者数据库只监听了特定网络地址。

Debian 可以检查服务状态：

sudo systemctl status mysql

还可以检查本机的 3306 端口监听情况：

sudo ss -lntp | grep 3306

如果 FastAPI 与 MySQL 位于同一台服务器，通常没有必要将 MySQL 的 3306 端口开放到公网。

3. ModuleNotFoundError

例如：

ModuleNotFoundError: No module named 'sqlalchemy'

这通常表示当前 Python 环境没有安装相应依赖，或者启动程序时没有使用正确的虚拟环境。

可以先激活虚拟环境，再安装：

python -m pip install -r requirements.txt

4. 数据库表不存在

请检查 models.py 中是否定义了模型，以及 main.py 是否导入了对应模型。

本教程使用 Base.metadata.create_all() 创建缺失的表，但它不会自动将已有表结构升级到最新版本。

正式项目应使用 Alembic 管理数据库迁移，避免直接依赖启动时自动建表来维护生产数据库结构。

十一、正式项目还需要完善哪些内容？

本教程实现了基础 CRUD，但距离生产级应用还有一些工作需要完成。

数据库迁移

使用 Alembic 管理表结构变更，记录每次数据库升级，避免手动修改数据库导致代码与表结构不一致。

身份认证与权限管理

根据业务需求引入登录认证、角色权限和资源访问控制。不能因为用户能够访问 API，就允许其随意修改或删除其他用户的数据。

统一异常处理

对数据库异常、业务异常和参数错误进行合理处理，避免向客户端暴露数据库连接信息或内部堆栈。

日志和监控

记录请求、异常和关键业务操作，便于排查线上问题。

分页与查询优化

根据实际数据规模增加过滤、排序、索引和分页策略。数据量较大时，还应评估深分页的性能问题。

配置与部署安全

通过环境变量或安全配置系统管理密钥；生产环境采用适当的进程管理、HTTPS、反向代理和访问控制。

十二、总结

通过本教程，我们完成了一个基于 FastAPI、MySQL 和 SQLAlchemy 2.0 的任务管理 API。

整个项目包含以下核心环节：

1. 创建 Python 虚拟环境并安装依赖。

2. 创建 MySQL 数据库和专用账户。

3. 使用 SQLAlchemy 定义 ORM 模型。

4. 使用 Pydantic 校验请求和响应数据。

5. 通过 FastAPI 实现增删改查。

6. 使用 Swagger UI 测试接口。

7. 了解数据库迁移、部署和安全方面的后续工作。

如果你已经掌握 Python 基础，这个项目可以作为后端开发的第一个数据库实战练习。