首页 / FastAPI 入门教程 / CRUD 操作

← 上一篇 下一篇 →
FastAPI 入门教程

CRUD 操作

本教程共 50 篇 · 第 34 篇 · 更新于 2026-08-12 · 约 10 分钟阅读

FastAPIFastAPI 入门教程SQLAlchemyCRUD增删改查

本节目标：掌握数据库最基本的四种操作——增（Create）、查（Read）、改（Update）、删（Delete），学会用 SQLAlchemy 的 Session 提交与回滚，并接到 FastAPI 接口上。

CRUD 是 Create、Read、Update、Delete 四个英文单词首字母。无论多复杂的系统，对数据的操作归根结底就是这四样。本章用「用户表」贯穿演示。

34-1 准备工作与依赖

先把第 33 章的引擎、基类、模型、会话工厂备好。这里补上 Pydantic 的响应模型，用于接口返回：

from typing import Annotated
from sqlalchemy import String, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, Session
from pydantic import BaseModel, ConfigDict
from fastapi import Depends, FastAPI

engine = create_engine(
    "sqlite:///./test.db",
    connect_args={"check_same_thread": False},
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    email: Mapped[str] = mapped_column(String(100), unique=True)

Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

# Pydantic v2：用 from_attributes 从 ORM 对象读属性（替代 v1 的 orm_mode）
class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    name: str
    email: str

app = FastAPI()

34-2 依赖：每个请求一个会话

FastAPI 推荐用「带 yield 的依赖」给每个请求发一个独立 Session，请求结束自动关闭，避免连接泄漏：

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 用一个类型别名，后面写起来更短
DbDep = Annotated[Session, Depends(get_db)]
Note

yield db 前面的代码在请求进来时执行，yield 后面的 finally 在响应返回后执行。这样保证每次请求都有干净的会话，且一定会被关掉。

34-3 增 Create：添加一条数据

新增数据的标准四步：建 ORM 对象 → add 进会话 → commit 提交 → refresh 刷新拿到数据库生成的值（如自增 id）。

class UserCreate(BaseModel):
    name: str
    email: str

@app.post("/users/", response_model=UserOut)
def create_user(user: UserCreate, db: DbDep):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)       # 记到笔记本
    db.commit()           # 真正写入数据库
    db.refresh(db_user)   # 把数据库生成的值（如 id）刷回对象
    return db_user        # Pydantic 用 from_attributes 转成 JSON

commit() 之前，数据只在会话里，数据库还不知道。refresh() 让对象拿到自增主键等数据库算出来的值。

Tip

想一次加多条，循环 add 后只 commit() 一次即可，比每条都提交快很多。

Note

上面这些接口用的是普通 def，不是 async def。原因：它们操作的是同步 Session，commit()、execute() 都是会阻塞线程的 I/O。同步 Session 必须配 def；只有把 Session 换成 AsyncSession 并用 await 时，才写成 async def（见 34-9）。本书默认推荐 async def，但遇到阻塞 I/O 就用 def，否则反而会在事件循环里卡住。

34-4 查 Read：查询与过滤

SQLAlchemy 2.0 推荐用 select() 构造查询，再交给 session.execute() 执行。result.scalars() 取出对象列表：

from sqlalchemy import select

@app.get("/users/", response_model=list[UserOut])
def read_users(db: DbDep, skip: int = 0, limit: int = 100):
    stmt = select(User).offset(skip).limit(limit)
    users = db.execute(stmt).scalars().all()
    return users

@app.get("/users/{user_id}", response_model=UserOut)
def read_user(user_id: int, db: DbDep):
    user = db.get(User, user_id)  # 按主键直接取，最简洁
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

带条件过滤时，用 where：

stmt = select(User).where(User.email == "a@x.com")
user = db.execute(stmt).scalars().first()
Note

db.get(User, id) 是 2.0 按主键查询的简便写法。复杂条件用 select(...).where(...)。不要用老教程里的 db.query(User).filter(...)（1.x 风格）。

34-5 改 Update：更新一条数据

更新分两种思路。简单场景：查出对象，直接改它的属性，再 commit。

@app.put("/users/{user_id}", response_model=UserOut)
def update_user(user_id: int, data: UserCreate, db: DbDep):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    user.name = data.name
    user.email = data.email
    db.commit()
    db.refresh(user)
    return user

更通用的写法是「只更新客户端传来的字段」。借助 Pydantic v2 的 model_dump(exclude_unset=True)，跳过没传的字段：

class UserUpdate(BaseModel):
    name: str | None = None
    email: str | None = None

@app.patch("/users/{user_id}", response_model=UserOut)
def patch_user(user_id: int, data: UserUpdate, db: DbDep):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    for key, value in data.model_dump(exclude_unset=True).items():
        setattr(user, key, value)
    db.commit()
    db.refresh(user)
    return user

exclude_unset=True 只取「客户端真正发来」的字段，没传的保持原值，这正是 PATCH 语义。

34-6 删 Delete：删除数据

删除用 db.delete(对象)，再 commit：

from fastapi import HTTPException

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: DbDep):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    db.delete(user)
    db.commit()
    return {"ok": True}
Tip

删除前务必先查对象。直接 delete(User).where(...) 也能批量删，但要小心别误删整张表。

34-7 提交与回滚：保证数据不出错

数据库操作讲究「事务」：一组操作要么全成功，要么全失败。SQLAlchemy 用 commit() 提交、rollback() 回滚。

session = SessionLocal()
try:
    session.add(User(name="小明", email="a@x.com"))
    session.add(User(name="小红", email="a@x.com"))
    session.commit()      # 两条一起生效
except Exception:
    session.rollback()    # 出错则全部撤销，库回到操作前状态
finally:
    session.close()

rollback() 在出错时把这次会话的改动全部丢弃，避免半成品数据残留在库里。

上面这个例子并非随手写的：两条记录用了同一个邮箱，而模型里 email 声明了 unique=True，所以 commit() 会因为唯一约束冲突而抛出 IntegrityError。此时第一条「小明」也不会被写进数据库，因为两次 add 处在同一个事务里，回滚是整体的。这正是事务的价值所在——你不必担心出现「只插了一半」的脏数据。

还要记住一点：会话一旦抛异常，就处于不可用状态，必须先 rollback() 才能继续执行后续查询，否则 SQLAlchemy 会持续报错提醒你事务尚未收尾。

34-8 with session.begin() 的简洁写法

每次手写 try/commit/finally 有点啰嗦。SQLAlchemy 提供了上下文管理器，自动提交或回滚：

with SessionLocal() as session:
    with session.begin():           # 退出时自动 commit，抛异常自动 rollback
        session.add(User(name="小明", email="a@x.com"))
# 退出外层 with 时，session 自动关闭

把两段 with 合并更紧凑：

with SessionLocal.begin() as session:
    session.add(User(name="小红", email="b@x.com"))
# 这里已经自动 commit 并关闭会话
Note

SessionLocal.begin() 同时管「事务」和「会话」，是日常写脚本、后台任务时最省心的写法。在 FastAPI 的 yield 依赖里，我们一般手动 commit，把提交时机留在路径操作函数内更灵活。

34-9 异步版 CRUD 长什么样

把 Session 换成 AsyncSession，操作前加 await，其余逻辑完全一致：

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

# 异步版 get_db 依赖（与 34-2 的同步 get_db 二选一使用）
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

async def create_user_async(name: str, email: str):
    async with AsyncSessionLocal() as session:
        user = User(name=name, email=email)
        session.add(user)
        await session.commit()
        await session.refresh(user)
        return user

# 在 async def 路径操作里
@app.post("/users/", response_model=UserOut)
async def create_user(user: UserCreate, db: Annotated[AsyncSession, Depends(get_db)]):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

异步查数据要 await db.execute(...)：

stmt = select(User)
result = await db.execute(stmt)
users = result.scalars().all()
Tip

异步不是「更快」，而是「不阻塞」。当数据库在远程、网络慢时，异步能让接口在等待数据库时去处理别的请求。本地 SQLite 场景用同步写法已经够用（关系在第 35 章、异步在第 36 章详讲）。

34-10 把数据库操作抽成独立函数

把增删改查直接写在路径操作函数里，代码很快会变长，接口一多就更乱。更好的做法是单独建一个 crud.py，把「和数据库打交道」的逻辑收拢成函数，路径操作只负责接收参数、调函数、返回结果。

# crud.py：专门放数据库操作，不依赖 FastAPI
from sqlalchemy import select
from sqlalchemy.orm import Session
from .models import User
from .schemas import UserCreate

def create_user(db: Session, data: UserCreate) -> User:
    user = User(name=data.name, email=data.email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def get_user(db: Session, user_id: int) -> User | None:
    return db.get(User, user_id)

def get_users(db: Session, skip: int = 0, limit: int = 100) -> list[User]:
    stmt = select(User).offset(skip).limit(limit)
    return db.execute(stmt).scalars().all()

路径操作就变得很干净：

@app.post("/users/", response_model=UserOut)
def create_user(user: UserCreate, db: DbDep):
    return crud.create_user(db=db, data=user)

这样做有三个好处。第一，路径操作变短，一眼看清接口在干嘛。第二，这些函数不依赖 FastAPI，能单独写单元测试，不用启动整个应用。第三，多个接口要查同一个东西时直接复用，不用复制粘贴。这是官方示例和真实项目都采用的套路。

Tip

小项目把 crud 函数写在同一个文件里也行；项目一大，按「用户、文章、订单」拆成多个模块，结构更清楚。

这一章你拿到了完整的增删改查能力。下一章我们让表与表之间产生关联——外键和关系，这才是关系型数据库真正的强项。

上一篇
连接数据库与定义模型
下一篇
表关系与外键