wwt｜逻辑与碎片

首页 / 知识库 / 0基础入门-阅读资料 / 0基础-后台入门

第五章 增删改查（CRUD）—— 后台的核心工作

本章目标：用 FastAPI + SQLModel 实现对数据库的增删改查，这是后台开发最核心的技能。

这是整个教程最重要的一章，请认真跟着敲代码。

5.1 什么是 CRUD

CRUD 是四个英文单词的首字母缩写：

字母英文中文HTTP 方法SQL 语句通俗理解
CCreate新增POSTINSERT往数据库里加一条新数据
RRead查询GETSELECT从数据库里找数据
UUpdate修改PUTUPDATE改数据库里已有的数据
DDelete删除DELETEDELETE从数据库里删一条数据

可以说，绝大多数后台系统的本质就是对数据的 CRUD。

  • 用户注册 → Create（新增一个用户）
  • 浏览商品 → Read（查询商品列表）
  • 修改密码 → Update（更新用户信息）
  • 注销账号 → Delete（删除用户数据）

接下来，我们围绕 Todo（待办事项）这个场景，逐个实现。

5.2 准备工作

确保你的项目中已经有这三个文件（上一章创建的）：

models.py（数据模型）

from sqlmodel import SQLModel, Field


class TodoCreate(SQLModel):
    title: str
    done: bool = False


class TodoUpdate(SQLModel):
    title: str | None = None
    done: bool | None = None


class Todo(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    done: bool = False

database.py（数据库配置）

from sqlmodel import SQLModel, Session, create_engine

DATABASE_URL = "sqlite:///database.db"

engine = create_engine(DATABASE_URL, echo=True)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


def get_session():
    with Session(engine) as session:
        yield session

准备好了，下面开始写 main.py。

5.3 Create —— 新增一条待办事项

接口设计

POST /todos
请求体：{"title": "买牛奶", "done": false}
响应：  {"id": 1, "title": "买牛奶", "done": false}

代码实现

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select
from contextlib import asynccontextmanager
from database import create_db_and_tables, get_session, engine
from models import Todo, TodoCreate, TodoUpdate


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(lifespan=lifespan)


@app.post("/todos", status_code=201)
def create_todo(todo: TodoCreate):
    with Session(engine) as session:
        db_todo = Todo.model_validate(todo)
        session.add(db_todo)
        session.commit()
        session.refresh(db_todo)
        return db_todo

逐行讲解

@app.post("/todos", status_code=201)
  • 定义一个 POST 接口，路径是 /todos
  • status_code=201 表示创建成功返回 201 状态码
def create_todo(todo: TodoCreate):
  • 参数 todo: TodoCreate 表示接收一个 JSON 请求体，格式按照 TodoCreate 模型
  • FastAPI 自动把 JSON 解析为 TodoCreate 对象
    with Session(engine) as session:
  • 打开一个数据库会话
        db_todo = Todo.model_validate(todo)
  • 把 TodoCreate 对象转换为 Todo 对象（带 id 字段的完整模型）
        session.add(db_todo)
  • 把这条数据添加到会话中（还没真正写入数据库）
        session.commit()
  • 提交事务，数据真正写入数据库
        session.refresh(db_todo)
  • 刷新对象，获取数据库自动生成的 id
        return db_todo
  • 返回完整的 Todo 对象（包含 id），FastAPI 自动转为 JSON

测试

启动服务后，打开 http://127.0.0.1:8000/docs：

  1. 找到 POST /todos
  2. 点击 “Try it out”
  3. 填写请求体：{"title": "买牛奶"}
  4. 点击 “Execute”
  5. 看到响应：{"id": 1, "title": "买牛奶", "done": false}

5.4 Read —— 查询待办事项

我们需要两个查询接口：查全部、查单条。

查询全部

GET /todos
响应：[{"id": 1, "title": "买牛奶", "done": false}, ...]
@app.get("/todos")
def get_todos():
    with Session(engine) as session:
        todos = session.exec(select(Todo)).all()
        return todos

讲解：

        todos = session.exec(select(Todo)).all()
  • select(Todo) → 构造一个查询语句：SELECT * FROM todo
  • session.exec(...) → 执行查询
  • .all() → 获取所有结果，返回一个列表

查询单条

GET /todos/1
响应：{"id": 1, "title": "买牛奶", "done": false}
@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")
        return todo

讲解：

        todo = session.get(Todo, todo_id)
  • session.get(Todo, todo_id) → 根据主键（id）查找一条数据
  • 如果找到了，返回 Todo 对象；如果没找到，返回 None
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")
  • 如果没找到，主动抛出 404 错误

测试

  1. 先用 POST /todos 添加几条数据
  2. GET /todos → 应该返回所有数据的列表
  3. GET /todos/1 → 返回 id 为 1 的数据
  4. GET /todos/999 → 返回 404 错误

5.5 Update —— 修改待办事项

接口设计

PUT /todos/1
请求体：{"done": true}           ← 只传要改的字段
响应：  {"id": 1, "title": "买牛奶", "done": true}

代码实现

@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, todo_update: TodoUpdate):
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")

        todo_data = todo_update.model_dump(exclude_unset=True)
        todo.sqlmodel_update(todo_data)

        session.add(todo)
        session.commit()
        session.refresh(todo)
        return todo

逐行讲解

def update_todo(todo_id: int, todo_update: TodoUpdate):
  • 接收两个参数：路径参数 todo_id 和请求体 todo_update
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")
  • 先查数据库，确认这条数据存在
        todo_data = todo_update.model_dump(exclude_unset=True)
  • model_dump() 把模型转成字典
  • exclude_unset=True 只保留用户传了的字段（没传的不改）
  • 比如用户只传了 {"done": true}，那 title 不会被改动
        todo.sqlmodel_update(todo_data)
  • 用传入的数据更新 todo 对象的字段
        session.add(todo)
        session.commit()
        session.refresh(todo)
        return todo
  • 保存到数据库，返回更新后的数据

测试

  1. 先确认有一条数据（GET /todos/1）
  2. PUT /todos/1，请求体：{"done": true}
  3. 再 GET /todos/1，检查 done 是否变成了 true

5.6 Delete —— 删除待办事项

接口设计

DELETE /todos/1
响应：{"message": "删除成功"}

代码实现

@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")

        session.delete(todo)
        session.commit()
        return {"message": "删除成功"}

讲解：

        session.delete(todo)
        session.commit()
  • session.delete(todo) → 标记要删除这条数据
  • session.commit() → 提交，真正从数据库中删除

测试

  1. 先确认有数据（GET /todos）
  2. DELETE /todos/1 → 返回 {"message": "删除成功"}
  3. 再 GET /todos → 列表中应该少了一条
  4. DELETE /todos/999 → 返回 404 错误

5.7 完整 main.py 代码

把所有接口放在一起，这就是完整的后台代码：

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from sqlmodel import Session, select
from database import create_db_and_tables, engine
from models import Todo, TodoCreate, TodoUpdate


@asynccontextmanager
async def lifespan(app: FastAPI):
    create_db_and_tables()
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/")
def index():
    return {"message": "Todo 后台服务已启动"}


# ==================== 增删改查接口 ====================


@app.post("/todos", status_code=201)
def create_todo(todo: TodoCreate):
    """新增一条待办事项"""
    with Session(engine) as session:
        db_todo = Todo.model_validate(todo)
        session.add(db_todo)
        session.commit()
        session.refresh(db_todo)
        return db_todo


@app.get("/todos")
def get_todos():
    """查询所有待办事项"""
    with Session(engine) as session:
        todos = session.exec(select(Todo)).all()
        return todos


@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    """查询单条待办事项"""
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")
        return todo


@app.put("/todos/{todo_id}")
def update_todo(todo_id: int, todo_update: TodoUpdate):
    """修改一条待办事项"""
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")

        todo_data = todo_update.model_dump(exclude_unset=True)
        todo.sqlmodel_update(todo_data)

        session.add(todo)
        session.commit()
        session.refresh(todo)
        return todo


@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    """删除一条待办事项"""
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo:
            raise HTTPException(status_code=404, detail="待办事项不存在")

        session.delete(todo)
        session.commit()
        return {"message": "删除成功"}

5.8 操作对照表

把所有操作对照一下，帮助记忆：

操作HTTP 方法URL函数数据库操作
新增POST/todoscreate_todosession.add() + commit()
查全部GET/todosget_todossession.exec(select(Todo)).all()
查单条GET/todos/{id}get_todosession.get(Todo, id)
修改PUT/todos/{id}update_todo修改字段 + commit()
删除DELETE/todos/{id}delete_todosession.delete() + commit()

数据库操作三步曲

不管是增、改、删，都遵循同样的套路：

1. 操作（add / 修改字段 / delete）
2. 提交（commit）
3. 刷新（refresh，如果需要返回最新数据）

本章小结

恭喜你！到这里你已经实现了一个完整的后台 CRUD 服务。

✅ Create  → POST   /todos      → 新增
✅ Read    → GET    /todos      → 查询全部
✅ Read    → GET    /todos/{id} → 查询单条
✅ Update  → PUT    /todos/{id} → 修改
✅ Delete  → DELETE /todos/{id} → 删除

这就是后台开发最核心的技能。市面上所有的后台管理系统、电商后台、内容管理系统……本质上都是更复杂的 CRUD。

下一章，我们将把前端和后端连起来——用 Vue 调用这些接口，实现完整的前后端联调。