FastAPI筑基_Day14_SQLAlchemy数据库实战

鲨鱼辣钊 2026-08-20 0 阅读12分钟

【FastAPI筑基-Day14】SQLAlchemy ORM数据库实战｜从零搭建MySQL连接、数据表、CRUD、项目解耦（最终工程化）

专栏：FastAPI零基础后端实战系列 标签：FastAPI、SQLAlchemy、MySQL、ORM、数据库CRUD、后端工程化 前置学习：Day11 路由分层与项目工程目录拆分、Day13 JWT动态令牌鉴权实战

---

一、前言

前面所有章节的数据全部都是内存假数据：服务一重启数据全丢，永远无法落地生产。Day14 正式接入数据库！

后端项目 90% 的业务都是围绕数据库 CRUD 展开，而 FastAPI 官方标配的数据库方案就是 SQLAlchemy ORM。本篇文章我们做一次完整的数据库实战：从零配置 MySQL 连接、ORM 模型映射自动建表、Session 会话管理，到 Model / Schema / CRUD / Router 标准四层分层，最终产出一个启动即跑、接口全通、数据落库的完整用户管理后端。

Day14 目标清单：

• ✅ SQLAlchemy 连接 MySQL 完整配置
• ✅ ORM 模型映射、自动建表
• ✅ 数据库会话 Session 管理
• ✅ 标准分层结构：Model / Schema / CRUD / Router
• ✅ 完整用户增删改查 CRUD
• ✅ 结合之前 JWT 登录、统一响应、全局异常

学完本章，你的项目就是一个完整上线级后端项目，本系列筑基篇也正式收官！

---

二、ORM 是什么？为什么要用 SQLAlchemy？

ORM（Object Relational Mapping，对象关系映射）：用 Python 类代替 SQL 语句，操作对象就是操作数据库。

打个比方：没有 ORM 时，你每操作一次数据库都要趴到窗口手写 SQL，像在国外办事大厅自己填外语表格；有了 ORM，相当于配了一位同声传译——你只管用 Python 操作对象（db.add(user)、db.query(User)），翻译官自动把你的操作翻译成 SQL 交给数据库执行，再把结果翻译回 Python 对象。

graph LR
    PY[Python 对象操作<br>db.add / db.query] --> ORM[SQLAlchemy ORM<br>对象关系映射]
    ORM --> SQL[自动生成 SQL]
    SQL --> DB[(MySQL 数据库)]
    DB --> ORM
    ORM --> PY

优势：

• 不用手写复杂 SQL，代码更优雅
• 自动防 SQL 注入（参数全部绑定传参，不拼接字符串）
• 适配多数据库（MySQL/SQLite/PostgreSQL）无缝切换，只改连接串
• 字段约束、类型、默认值统一在模型里管理

---

三、安装依赖与 MySQL 准备

pip install sqlalchemy pymysql "passlib[bcrypt]" "bcrypt==4.0.1"

说明：

• pymysql：MySQL 驱动，负责真正和 MySQL 建立 TCP 连接
• sqlalchemy：ORM 核心库
• passlib + bcrypt：密码哈希加密（Day13 同款），数据库绝不存明文密码

MySQL 前置准备：

1. 本地安装并启动 MySQL
2. 创建数据库：

CREATE DATABASE fastapi_db DEFAULT CHARACTER SET utf8mb4;

1. 后面 database.py 中的账号密码修改为自己的（本文为本地学习环境使用 root/123456，生产请务必换强密码）

---

四、企业级标准四层架构（重点）

我们将项目彻底解耦，这是大厂 FastAPI 项目的标准结构：

层	职责	禁止事项
models	数据库实体模型，一个类 = 一张表	写业务逻辑
schemas	Pydantic 请求/响应模型，参数校验、序列化	碰数据库
crud	æ•°æ®åº“æ“ä½œé€»è¾‘ï¼Œå¢žåˆ æ”¹æŸ¥å ¨åœ¨è¿™	碰 HTTP、碰参数校验
routers	接口路由层，只负责接收参数、调用 CRUD、返回响应	直接写 SQL

一句话：路由不碰数据库、CRUD 不做参数校验、模型各司其职。

请求的完整流转：

graph TD
    Client[客户端请求] --> Router[routers 路由层]
    Router -. Pydantic 自动校验参数 .-> Schema[schemas 校验层]
    Router --> CRUD[crud 数据操作层]
    CRUD --> Model[models 模型层]
    Model --> DB[(MySQL)]
    CRUD --> Router
    Router --> Client

---

五、完整项目目录

fastapi_project/
├── main.py                 # 项目入口
├── database.py             # 数据库连接配置
├── requirements.txt        # 依赖清单
├── models/                 # 数据库表模型
│   └── user.py
├── schemas/                # 数据校验模型
│   └── user.py
├── crud/                   # 数据库操作
│   └── user.py
├── routers/                # 接口路由
│   └── user.py
└── common/                 # 公共工具
    └── response.py

---

六、分步核心代码实现

版本提醒：网上很多老教程用 from sqlalchemy.ext.declarative import declarative_base、sessionmaker(autocommit=False)、Pydantic 的 orm_mode = True。这些在 SQLAlchemy 2.x / Pydantic v2 下已移除或废弃，直接抄会报错。本章代码全部按新版写法编写，可直接运行。

1. database.py 数据库全局配置

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# 数据库连接地址，请修改为自己的账号密码
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/fastapi_db?charset=utf8mb4"

# 创建引擎（连接池）
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_pre_ping=True,   # 使用前先 ping 一下，避免使用到超时断开的废连接
    pool_recycle=3600,    # 连接每小时回收重建，防止数据库主动断开
)

# 会话工厂：每次请求通过它创建一个独立 Session
SessionLocal = sessionmaker(autoflush=False, bind=engine)

# 所有 ORM 模型的基类
Base = declarative_base()


# FastAPI 依赖注入：每个请求拿到独立会话，请求结束自动关闭
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

连接串格式：mysql+pymysql://账号:密码@主机:端口/数据库名?charset=utf8mb4。get_db 是 Day12 依赖注入的实战应用：用 yield 把会话交给接口，请求结束 finally 保证关闭，永不泄漏连接。

2. models/user.py 数据库表模型

from sqlalchemy import Column, Integer, String, Boolean
from database import Base


class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    username = Column(String(50), unique=True, nullable=False, index=True)
    password = Column(String(128), nullable=False)
    email = Column(String(100), nullable=True)
    is_active = Column(Boolean, default=True)

一个类 = 一张表，一个 Column = 一列。unique 唯一约束、nullable 非空约束、index 加索引、default 默认值，全部在模型里声明。

3. schemas/user.py 请求响应模型

from typing import Optional
from pydantic import BaseModel, ConfigDict, Field


# 创建用户：请求体校验
class UserCreate(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    password: str = Field(min_length=6, max_length=32)
    email: Optional[str] = None


# 更新用户：所有字段可选，传哪个改哪个
class UserUpdate(BaseModel):
    password: Optional[str] = Field(default=None, min_length=6, max_length=32)
    email: Optional[str] = None
    is_active: Optional[bool] = None


# 用户信息返回：不含密码字段
class UserInfo(BaseModel):
    id: int
    username: str
    email: Optional[str]
    is_active: bool

    # SQLAlchemy 2.x + Pydantic v2 写法，等价于旧版 orm_mode = True
    model_config = ConfigDict(from_attributes=True)

注意 UserInfo 里没有 password 字段：数据库对象转响应时密码自动被过滤，绝不怕手滑把密码吐给前端。

4. crud/user.py 数据库CRUD逻辑

from passlib.context import CryptContext
from sqlalchemy.orm import Session
from models.user import User
from schemas.user import UserCreate, UserUpdate

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


# 密码加密
def get_hash_pwd(password: str):
    return pwd_context.hash(password)


# 创建用户
def create_user(db: Session, user: UserCreate):
    db_user = User(
        username=user.username,
        password=get_hash_pwd(user.password),
        email=user.email
    )
    db.add(db_user)      # 加入会话
    db.commit()          # 提交事务，真正写库
    db.refresh(db_user)  # 刷新拿到自增 id 等数据库回填字段
    return db_user


# 根据用户名查询用户
def get_user_by_name(db: Session, username: str):
    return db.query(User).filter(User.username == username).first()


# 根据ID查询用户
def get_user_by_id(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()


# 获取用户列表（分页）
def get_user_list(db: Session, page: int = 1, size: int = 10):
    offset = (page - 1) * size
    return db.query(User).offset(offset).limit(size).all()


# 更新用户：只更新请求里实际携带的字段
def update_user(db: Session, user_id: int, user: UserUpdate):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        return None
    data = user.model_dump(exclude_unset=True)
    if "password" in data:
        data["password"] = get_hash_pwd(data["password"])
    for key, value in data.items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return db_user


# 删除用户
def delete_user(db: Session, user_id: int) -> bool:
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        return False
    db.delete(db_user)
    db.commit()
    return True

CRUD 层只认 Session 和模型，完全不知道 HTTP 的存在，换到命令行脚本、定时任务里也能直接复用。

5. routers/user.py 接口路由层

from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from database import get_db
from schemas.user import UserCreate, UserInfo, UserUpdate
from crud import user as user_crud
from common.response import success_response, fail_response

router = APIRouter(prefix="/user", tags=["用户数据库模块"])


# 创建用户
@router.post("/add", summary="新增用户")
def add_user(user: UserCreate, db: Session = Depends(get_db)):
    if user_crud.get_user_by_name(db, user.username):
        return fail_response(msg="用户名已存在")
    res = user_crud.create_user(db, user)
    return success_response(data=res.id, msg="用户创建成功")


# 根据用户名查询
@router.get("/name", summary="根据用户名查询")
def get_user(name: str, db: Session = Depends(get_db)):
    res = user_crud.get_user_by_name(db, name)
    if not res:
        return fail_response(code=404, msg="用户不存在")
    return success_response(data=UserInfo.model_validate(res))


# 用户列表分页
@router.get("/list", summary="用户列表分页")
def user_list(
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100),
    db: Session = Depends(get_db)
):
    data = user_crud.get_user_list(db, page, size)
    return success_response(data=[UserInfo.model_validate(u) for u in data])


# 根据ID更新用户
@router.put("/{user_id}", summary="更新用户")
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    res = user_crud.update_user(db, user_id, user)
    if not res:
        return fail_response(code=404, msg="用户不存在")
    return success_response(data=UserInfo.model_validate(res), msg="更新成功")


# 根据ID删除用户
@router.delete("/{user_id}", summary="删除用户")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    ok = user_crud.delete_user(db, user_id)
    if not ok:
        return fail_response(code=404, msg="用户不存在")
    return success_response(msg="删除成功")

路由层全程没有一行 SQL：Depends(get_db) 注入会话 → 调 CRUD → 套统一响应，三件事而已。

6. common/response.py 统一响应

from typing import Any
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse


def success_response(data: Any = None, msg: str = "请求成功") -> JSONResponse:
    return JSONResponse(status_code=200, content={"code": 200, "msg": msg, "data": jsonable_encoder(data)})


def fail_response(code: int = 400, msg: str = "请求失败", data: Any = None) -> JSONResponse:
    return JSONResponse(status_code=200, content={"code": code, "msg": msg, "data": jsonable_encoder(data)})

jsonable_encoder 负责把 Pydantic 模型等对象转成可 JSON 序列化的结构，Day10 的老朋友。

7. main.py 主入口 & 自动建表

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from database import engine, Base
from models.user import User  # noqa: F401 建表前必须先导入模型完成注册
from routers.user import router as user_router

# 自动创建数据表（表不存在时才创建）
Base.metadata.create_all(bind=engine)

app = FastAPI(title="Day14 SQLAlchemy数据库实战")

# 跨域
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(user_router)


@app.get("/")
def root():
    return {"msg": "数据库项目启动成功！访问 /docs 调试接口"}


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Base.metadata.create_all(bind=engine) 会扫描所有继承 Base 的模型，启动时自动建表，无需手写建表 SQL。注意必须先导入模型模块，模型注册到 Base 后才会被建表扫到。

---

七、启动项目，全流程实测 CRUD

启动项目：

python main.py

1. 确认自动建表

打开 MySQL 客户端查看，user 表已经自动创建：

mysql> SHOW TABLES;
+----------------------+
| Tables_in_fastapi_db |
+----------------------+
| user                 |
+----------------------+

2. Swagger 文档一览

访问 /docs，五个用户接口整齐挂在「用户数据库模块」下：

3. 增：新增两个用户 + 重复新增拦截

curl -X POST http://localhost:8000/user/add -H "Content-Type: application/json" \
  -d '{"username":"zhangsan","password":"123456","email":"zhangsan@example.com"}'

{"code":200,"msg":"用户创建成功","data":1}

curl -X POST http://localhost:8000/user/add -H "Content-Type: application/json" \
  -d '{"username":"lisi","password":"abc123456"}'

{"code":200,"msg":"用户创建成功","data":2}

重复新增 zhangsan，被路由层查重拦截：

{"code":400,"msg":"用户名已存在","data":null}

4. 查：按用户名查询 + 列表分页

curl "http://localhost:8000/user/name?name=zhangsan"

{"code":200,"msg":"请求成功","data":{"id":1,"username":"zhangsan","email":"zhangsan@example.com","is_active":true}}

返回的是 UserInfo 结构：只有 id、username、email、is_active，密码字段被自动过滤。查询不存在的用户：

{"code":404,"msg":"用户不存在","data":null}

分页列表：

curl "http://localhost:8000/user/list?page=1&size=10"

{"code":200,"msg":"请求成功","data":[{"id":1,"username":"zhangsan","email":"zhangsan@example.com","is_active":true},{"id":2,"username":"lisi","email":null,"is_active":true}]}

5. 改：更新用户邮箱

curl -X PUT http://localhost:8000/user/1 -H "Content-Type: application/json" \
  -d '{"email":"zhangsan_new@example.com"}'

{"code":200,"msg":"更新成功","data":{"id":1,"username":"zhangsan","email":"zhangsan_new@example.com","is_active":true}}

UserUpdate 所有字段可选，传哪个改哪个，不传的字段原样保留。

6. 删：删除用户并确认

curl -X DELETE http://localhost:8000/user/2

{"code":200,"msg":"删除成功","data":null}

删除后再查 lisi：

{"code":404,"msg":"用户不存在","data":null}

7. 参数校验照旧生效

密码太短，Pydantic 直接 422 拦下，请求根本进不到数据库层：

curl -X POST http://localhost:8000/user/add -H "Content-Type: application/json" \
  -d '{"username":"wangwu","password":"123"}'

{"detail":[{"type":"string_too_short","loc":["body","password"],"msg":"String should have at least 6 characters","input":"123","ctx":{"min_length":6}}]}

8. 在 Swagger 里在线新增一个用户

在 /docs 里展开 POST /user/add，点 Try it out 填入请求体执行，响应 code 200，data 返回新用户 id：

9. 最后去 MySQL 里亲眼确认

mysql> SELECT id, username, LEFT(password, 13) AS pwd_prefix, email FROM user;
+----+----------+---------------+---------------------------+
| id | username | pwd_prefix    | email                     |
+----+----------+---------------+---------------------------+
|  1 | zhangsan | $2b$12$ukg1Jt | zhangsan_new@example.com  |
|  3 | wangwu   | $2b$12$QftetQ | wangwu@example.com        |
+----+----------+---------------+---------------------------+

数据真的落在 MySQL 里了：重启服务数据依然在；密码列存的是 $2b$12$... 开头的 bcrypt 哈希，不是明文。至此 CRUD 全链路闭环。

---

八、结合之前章节：JWT 鉴权与全局异常（选学）

统一响应本章已经用上，JWT 和全局异常在 Day13、Day10 都已写好，接入只需各加一行：

挂 JWT 鉴权：把 Day13 的 get_current_user 原样抄到 common/auth.py，然后给路由加一个 dependencies：

from fastapi import Depends
from common.auth import get_current_user

# 该模块下所有接口都必须登录后才能访问
router = APIRouter(prefix="/user", tags=["用户数据库模块"],
                   dependencies=[Depends(get_current_user)])

挂全局异常：把 Day10 的两个 @app.exception_handler 注册到 main.py，数据库报错、参数报错统统被统一响应格式兜底。具体实现回看 Day10 / Day13，本章不重复贴代码。

---

九、核心知识点总结（面试必背）

概念	一句话说明
Engine	æ•°æ®åº“è¿žæŽ¥å¼•æ“Žï¼Œå ¨å±€å”¯ä¸€ï¼Œç®¡ç†è¿žæŽ¥æ±
Session	æ•°æ®åº“ä¼šè¯ï¼Œæ¯æ¬¡è¯·æ±‚ç‹¬ç«‹ä¼šè¯ã€è¯·æ±‚ç»“æŸè‡ªåŠ¨å ³é—­
Base 模型	所有数据库表的父类，create_all 自动建表
四层架构	路由、模型、校验、CRUD å®Œå ¨åˆ†ç¦»ï¼Œå„å¸å ¶èŒ
from_attributes	数据库 ORM 对象自动转 Pydantic 模型返回前端（旧版叫 orm_mode=True）
add/commit/refresh	åŠ å ¥ä¼šè¯ → 提交事务 → 刷新拿数据库回填字段，写库三步曲

---

十、本系列完整收官总结（Day1～Day14）

到此，FastAPI 零基础筑基系列完整完结！你从零掌握了：

• Python 异步核心、aiohttp 高并发
• 工程日志、命令行参数、代码规范
• Pydantic 全自动参数校验
• GET/POST、三大传参、高级参数校验
• 跨域、静态文件、全局异常、统一响应
• 路由分层、模块化大型项目架构
• Depends 依赖注入、接口权限控制
• JWT 无状态登录鉴权体系
• SQLAlchemy ORM 数据库 CRUD

已完全具备独立开发企业级前后端分离项目的能力！

---

十一、后续进阶预告

后续进阶系列：FastAPI 完整前后端分离实战项目、Redis 缓存、定时任务、日志持久化、Docker 部署、线上服务器上线。我们进阶篇见！