第8章 - CRUD 操作

嗨，朋友！我是长安。

这一章我们来实现完整的 CRUD 操作。说实话，这是整个教程中非常重要的一章！掌握了 CRUD，你就掌握了后端开发的核心。我当年就是通过这个掌握了 FastAPI 的精髓。

🎯 本章目标

理解什么是 CRUD
学会实现增删改查四种操作
掌握 RESTful API 设计规范
完成一个完整的 CRUD 示例

1️⃣ 什么是 CRUD？

CRUD 是四种基本数据操作的缩写，几乎所有的应用都离不开这四种操作！

操作	英文	HTTP 方法	示例
创建	Create	POST	创建新用户
读取	Read	GET	获取用户信息
更新	Update	PUT/PATCH	修改用户信息
删除	Delete	DELETE	删除用户

几乎所有的应用都离不开这四种操作！

2️⃣ RESTful API 设计规范

RESTful 是一种 API 设计风格，主要规范：

URL 设计

GET    /users          # 获取用户列表
GET    /users/{id}     # 获取单个用户
POST   /users          # 创建用户
PUT    /users/{id}     # 更新用户（全量）
PATCH  /users/{id}     # 更新用户（部分）
DELETE /users/{id}     # 删除用户

命名规范

使用名词，不用动词：/users ✅，/getUsers ❌
使用复数形式：/users ✅，/user ❌
使用小写字母：/users ✅，/Users ❌
使用连字符：/user-profiles ✅，/user_profiles ⚠️

3️⃣ 实现 CRUD - 准备工作

首先，定义数据模型：

from typing import List, Optional
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(title="CRUD 示例")

# ========== 数据模型 ==========

class BookBase(BaseModel):
    """图书基础模型"""
    title: str = Field(min_length=1, max_length=100, description="书名")
    author: str = Field(min_length=1, max_length=50, description="作者")
    price: float = Field(gt=0, description="价格")
    description: Optional[str] = Field(default=None, max_length=500, description="简介")

class BookCreate(BookBase):
    """创建图书请求模型"""
    pass

class BookUpdate(BaseModel):
    """更新图书请求模型（所有字段可选）"""
    title: Optional[str] = Field(default=None, min_length=1, max_length=100)
    author: Optional[str] = Field(default=None, min_length=1, max_length=50)
    price: Optional[float] = Field(default=None, gt=0)
    description: Optional[str] = Field(default=None, max_length=500)

class Book(BookBase):
    """图书响应模型"""
    id: int

# ========== 模拟数据库 ==========

fake_db: dict[int, dict] = {
    1: {"id": 1, "title": "Python入门", "author": "张三", "price": 59.0, "description": "Python基础教程"},
    2: {"id": 2, "title": "FastAPI实战", "author": "李四", "price": 79.0, "description": "FastAPI开发指南"},
    3: {"id": 3, "title": "数据结构", "author": "王五", "price": 49.0, "description": None},
}
id_counter = 3

4️⃣ Create - 创建

@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED, tags=["图书管理"])
def create_book(book: BookCreate):
    """
    创建新图书
    
    - **title**: 书名（必填）
    - **author**: 作者（必填）
    - **price**: 价格（必填，大于0）
    - **description**: 简介（可选）
    """
    global id_counter
    id_counter += 1
    
    new_book = {
        "id": id_counter,
        **book.model_dump()
    }
    fake_db[id_counter] = new_book
    
    return new_book

测试创建

POST /books
Content-Type: application/json

{
    "title": "Vue.js入门",
    "author": "赵六",
    "price": 69.0,
    "description": "Vue.js前端开发"
}

响应：

{
    "id": 4,
    "title": "Vue.js入门",
    "author": "赵六",
    "price": 69.0,
    "description": "Vue.js前端开发"
}

5️⃣ Read - 读取

获取列表

@app.get("/books", response_model=List[Book], tags=["图书管理"])
def get_books(
    skip: int = 0,
    limit: int = 10,
    keyword: Optional[str] = None
):
    """
    获取图书列表
    
    - **skip**: 跳过数量（分页用）
    - **limit**: 返回数量（分页用）
    - **keyword**: 搜索关键词（可选）
    """
    books = list(fake_db.values())
    
    # 关键词搜索
    if keyword:
        books = [
            book for book in books 
            if keyword.lower() in book["title"].lower() 
            or keyword.lower() in book["author"].lower()
        ]
    
    # 分页
    return books[skip:skip + limit]

获取单个

@app.get("/books/{book_id}", response_model=Book, tags=["图书管理"])
def get_book(book_id: int):
    """根据ID获取图书详情"""
    if book_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"图书 {book_id} 不存在"
        )
    return fake_db[book_id]

6️⃣ Update - 更新

PUT - 全量更新

@app.put("/books/{book_id}", response_model=Book, tags=["图书管理"])
def update_book(book_id: int, book: BookCreate):
    """
    全量更新图书信息
    
    需要提供所有字段
    """
    if book_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"图书 {book_id} 不存在"
        )
    
    updated_book = {
        "id": book_id,
        **book.model_dump()
    }
    fake_db[book_id] = updated_book
    
    return updated_book

PATCH - 部分更新

@app.patch("/books/{book_id}", response_model=Book, tags=["图书管理"])
def partial_update_book(book_id: int, book: BookUpdate):
    """
    部分更新图书信息
    
    只需要提供要更新的字段
    """
    if book_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"图书 {book_id} 不存在"
        )
    
    stored_book = fake_db[book_id]
    update_data = book.model_dump(exclude_unset=True)  # 只获取设置了的字段
    
    for field, value in update_data.items():
        stored_book[field] = value
    
    return stored_book

PUT vs PATCH

方法	说明	示例
PUT	全量更新，需要提供所有字段	更新整个用户信息
PATCH	部分更新，只提供要修改的字段	只更新用户名

# PUT 请求 - 需要所有字段
PUT /books/1
{
    "title": "Python入门（第2版）",
    "author": "张三",
    "price": 69.0,
    "description": "Python基础教程"
}

# PATCH 请求 - 只需要修改的字段
PATCH /books/1
{
    "price": 69.0
}

7️⃣ Delete - 删除

@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["图书管理"])
def delete_book(book_id: int):
    """删除图书"""
    if book_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"图书 {book_id} 不存在"
        )
    
    del fake_db[book_id]
    return None  # 204 No Content

8️⃣ 完整示例

from typing import List, Optional
from fastapi import FastAPI, HTTPException, status, Query
from pydantic import BaseModel, Field

app = FastAPI(
    title="图书管理 API",
    description="一个完整的 CRUD 示例",
    version="1.0.0"
)

# ========== 数据模型 ==========

class BookBase(BaseModel):
    title: str = Field(min_length=1, max_length=100, description="书名")
    author: str = Field(min_length=1, max_length=50, description="作者")
    price: float = Field(gt=0, description="价格")
    description: Optional[str] = Field(default=None, max_length=500, description="简介")

class BookCreate(BookBase):
    pass

class BookUpdate(BaseModel):
    title: Optional[str] = Field(default=None, min_length=1, max_length=100)
    author: Optional[str] = Field(default=None, min_length=1, max_length=50)
    price: Optional[float] = Field(default=None, gt=0)
    description: Optional[str] = Field(default=None, max_length=500)

class Book(BookBase):
    id: int

class BookListResponse(BaseModel):
    total: int
    items: List[Book]

# ========== 模拟数据库 ==========

fake_db: dict[int, dict] = {
    1: {"id": 1, "title": "Python入门", "author": "张三", "price": 59.0, "description": "Python基础教程"},
    2: {"id": 2, "title": "FastAPI实战", "author": "李四", "price": 79.0, "description": "FastAPI开发指南"},
    3: {"id": 3, "title": "数据结构", "author": "王五", "price": 49.0, "description": None},
}
id_counter = 3

# ========== CRUD 接口 ==========

# Create
@app.post("/books", response_model=Book, status_code=status.HTTP_201_CREATED, tags=["图书管理"])
def create_book(book: BookCreate):
    """创建新图书"""
    global id_counter
    id_counter += 1
    new_book = {"id": id_counter, **book.model_dump()}
    fake_db[id_counter] = new_book
    return new_book

# Read - 列表
@app.get("/books", response_model=BookListResponse, tags=["图书管理"])
def get_books(
    page: int = Query(default=1, ge=1, description="页码"),
    size: int = Query(default=10, ge=1, le=100, description="每页数量"),
    keyword: Optional[str] = Query(default=None, description="搜索关键词"),
    min_price: Optional[float] = Query(default=None, ge=0, description="最低价格"),
    max_price: Optional[float] = Query(default=None, ge=0, description="最高价格"),
):
    """获取图书列表，支持分页和搜索"""
    books = list(fake_db.values())
    
    # 关键词搜索
    if keyword:
        books = [b for b in books if keyword.lower() in b["title"].lower() or keyword.lower() in b["author"].lower()]
    
    # 价格筛选
    if min_price is not None:
        books = [b for b in books if b["price"] >= min_price]
    if max_price is not None:
        books = [b for b in books if b["price"] <= max_price]
    
    total = len(books)
    start = (page - 1) * size
    items = books[start:start + size]
    
    return BookListResponse(total=total, items=items)

# Read - 单个
@app.get("/books/{book_id}", response_model=Book, tags=["图书管理"])
def get_book(book_id: int):
    """获取图书详情"""
    if book_id not in fake_db:
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
    return fake_db[book_id]

# Update - 全量
@app.put("/books/{book_id}", response_model=Book, tags=["图书管理"])
def update_book(book_id: int, book: BookCreate):
    """全量更新图书"""
    if book_id not in fake_db:
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
    updated_book = {"id": book_id, **book.model_dump()}
    fake_db[book_id] = updated_book
    return updated_book

# Update - 部分
@app.patch("/books/{book_id}", response_model=Book, tags=["图书管理"])
def partial_update_book(book_id: int, book: BookUpdate):
    """部分更新图书"""
    if book_id not in fake_db:
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
    
    stored_book = fake_db[book_id]
    update_data = book.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        stored_book[field] = value
    return stored_book

# Delete
@app.delete("/books/{book_id}", status_code=status.HTTP_204_NO_CONTENT, tags=["图书管理"])
def delete_book(book_id: int):
    """删除图书"""
    if book_id not in fake_db:
        raise HTTPException(status_code=404, detail=f"图书 {book_id} 不存在")
    del fake_db[book_id]
    return None

# 额外接口：批量删除
@app.post("/books/batch-delete", tags=["图书管理"])
def batch_delete_books(ids: List[int]):
    """批量删除图书"""
    deleted = []
    not_found = []
    for book_id in ids:
        if book_id in fake_db:
            del fake_db[book_id]
            deleted.append(book_id)
        else:
            not_found.append(book_id)
    return {
        "deleted": deleted,
        "not_found": not_found,
        "message": f"成功删除 {len(deleted)} 本图书"
    }

9️⃣ 错误处理

使用 HTTPException

from fastapi import HTTPException, status

@app.get("/books/{book_id}")
def get_book(book_id: int):
    if book_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="图书不存在",
            headers={"X-Error": "Book not found"}  # 可选：自定义响应头
        )
    return fake_db[book_id]

常见错误场景

# 资源不存在
raise HTTPException(status_code=404, detail="资源不存在")

# 参数错误
raise HTTPException(status_code=400, detail="参数错误")

# 未授权
raise HTTPException(status_code=401, detail="请先登录")

# 禁止访问
raise HTTPException(status_code=403, detail="没有权限")

# 冲突（如重复创建）
raise HTTPException(status_code=409, detail="资源已存在")

📝 小结

本章我们学习了：

✅ CRUD 的概念和对应的 HTTP 方法
✅ RESTful API 设计规范
✅ 实现创建、读取、更新、删除操作
✅ PUT 和 PATCH 的区别
✅ 错误处理

🏃 下一步

目前我们的数据都存在内存中，程序重启就没了。下一章学习如何连接数据库，实现数据持久化！

👉 第9章 - 数据库操作

💪 练习题

实现一个用户管理的 CRUD API，包含：用户名、邮箱、年龄、状态
添加搜索功能：按用户名搜索
添加批量操作：批量创建用户、批量删除用户

参考答案

from typing import List, Optional
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI()

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    email: str
    age: int = Field(ge=0, le=150)
    is_active: bool = True

class UserUpdate(BaseModel):
    username: Optional[str] = None
    email: Optional[str] = None
    age: Optional[int] = None
    is_active: Optional[bool] = None

class User(UserCreate):
    id: int

fake_db = {}
id_counter = 0

# Create
@app.post("/users", response_model=User, status_code=201)
def create_user(user: UserCreate):
    global id_counter
    id_counter += 1
    new_user = {"id": id_counter, **user.model_dump()}
    fake_db[id_counter] = new_user
    return new_user

# Read - 列表（带搜索）
@app.get("/users", response_model=List[User])
def get_users(username: Optional[str] = None):
    users = list(fake_db.values())
    if username:
        users = [u for u in users if username.lower() in u["username"].lower()]
    return users

# Read - 单个
@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return fake_db[user_id]

# Update
@app.patch("/users/{user_id}", response_model=User)
def update_user(user_id: int, user: UserUpdate):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    stored = fake_db[user_id]
    for k, v in user.model_dump(exclude_unset=True).items():
        stored[k] = v
    return stored

# Delete
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
    if user_id not in fake_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    del fake_db[user_id]

# 批量创建
@app.post("/users/batch", response_model=List[User])
def batch_create_users(users: List[UserCreate]):
    global id_counter
    created = []
    for user in users:
        id_counter += 1
        new_user = {"id": id_counter, **user.model_dump()}
        fake_db[id_counter] = new_user
        created.append(new_user)
    return created

# 批量删除
@app.post("/users/batch-delete")
def batch_delete_users(ids: List[int]):
    deleted = [i for i in ids if i in fake_db]
    for i in deleted:
        del fake_db[i]
    return {"deleted": deleted}