4.3 用户接口实战
本章目标：完成用户资源的完整 CRUD 接口，打通 schemas → crud → router 的完整链路。这是全教程的核心章节，请务必逐行理解并亲手敲完。
1. app/schemas.py（用户部分）
先定义接口的输入输出结构：
from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field
# ---------- 用户 ----------
class UserCreate(BaseModel):
"""注册时接收的数据"""
username: str = Field(min_length=3, max_length=20)
email: EmailStr
password: str = Field(min_length=6, max_length=64)
class UserUpdate(BaseModel):
"""更新时接收的数据：全部可选"""
username: str | None = Field(default=None, min_length=3, max_length=20)
email: EmailStr | None = None
class UserOut(BaseModel):
"""返回给前端的用户数据（无密码）"""
model_config = ConfigDict(from_attributes=True)
id: int
username: str
email: EmailStr
is_active: bool
created_at: datetime
EmailStr需要pip install "pydantic[email]"（fastapi[standard] 已附带）。
关键配置：from_attributes
model_config = ConfigDict(from_attributes=True) 是 Pydantic 与 SQLAlchemy 的桥梁。
没有它，Pydantic 只能从字典创建模型；加上它，Pydantic 就能从任意带属性的对象（比如 SQLAlchemy 的 User 实例）读取字段。这样接口函数直接 return 数据库对象，FastAPI 就会按 UserOut 自动提取字段、序列化成 JSON。
旧教程里的
class Config: orm_mode = True是 Pydantic v1 的写法，作用相同。
2. app/crud.py（用户部分）
把所有数据库操作集中在这一层：
from sqlalchemy import select
from sqlalchemy.orm import Session
from app import models, schemas
# ---------- 密码哈希（临时简化版！第五部分换成 bcrypt 真实现） ----------
def hash_password(password: str) -> str:
return f"fake_hashed_{password}"
# ---------- 用户 ----------
def get_user(db: Session, user_id: int) -> models.User | None:
return db.get(models.User, user_id)
def get_user_by_username(db: Session, username: str) -> models.User | None:
return db.scalars(
select(models.User).where(models.User.username == username)
).first()
def get_user_by_email(db: Session, email: str) -> models.User | None:
return db.scalars(
select(models.User).where(models.User.email == email)
).first()
def list_users(db: Session, offset: int = 0, limit: int = 10) -> list[models.User]:
stmt = select(models.User).order_by(models.User.id).offset(offset).limit(limit)
return list(db.scalars(stmt).all())
def create_user(db: Session, user_in: schemas.UserCreate) -> models.User:
user = models.User(
username=user_in.username,
email=user_in.email,
hashed_password=hash_password(user_in.password),
)
db.add(user)
db.commit()
db.refresh(user) # 从数据库重新读取，拿到 id、created_at 等生成值
return user
def update_user(
db: Session, user: models.User, user_in: schemas.UserUpdate
) -> models.User:
# 只更新调用方真正传了的字段
for field, value in user_in.model_dump(exclude_unset=True).items():
setattr(user, field, value)
db.commit()
db.refresh(user)
return user
def delete_user(db: Session, user: models.User) -> None:
db.delete(user)
db.commit()
要点：
- crud 函数只做数据库操作，不抛 HTTPException——404 之类的判断留给路由层。这样 crud 层保持纯粹，将来写测试、写后台脚本都能复用
db.refresh(user)：commit 后从数据库把该行重新读一遍，确保created_at这类数据库生成的值填进对象update_user用了第二部分讲的exclude_unset部分更新模式
3. app/routers/users.py
from fastapi import APIRouter, HTTPException, status
from app import crud, schemas
from app.database import SessionDep
router = APIRouter(prefix="/users", tags=["用户"])
@router.post("", response_model=schemas.UserOut, status_code=status.HTTP_201_CREATED)
def create_user(user_in: schemas.UserCreate, db: SessionDep):
"""注册新用户"""
if crud.get_user_by_username(db, user_in.username):
raise HTTPException(status_code=409, detail="用户名已存在")
if crud.get_user_by_email(db, user_in.email):
raise HTTPException(status_code=409, detail="邮箱已被注册")
return crud.create_user(db, user_in)
@router.get("", response_model=list[schemas.UserOut])
def list_users(db: SessionDep, page: int = 1, size: int = 10):
"""用户列表（分页）"""
size = min(size, 100)
return crud.list_users(db, offset=(page - 1) * size, limit=size)
@router.get("/{user_id}", response_model=schemas.UserOut)
def get_user(user_id: int, db: SessionDep):
"""查询单个用户"""
user = crud.get_user(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
return user
@router.patch("/{user_id}", response_model=schemas.UserOut)
def update_user(user_id: int, user_in: schemas.UserUpdate, db: SessionDep):
"""更新用户信息（只改传了的字段）"""
user = crud.get_user(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
if user_in.username and user_in.username != user.username:
if crud.get_user_by_username(db, user_in.username):
raise HTTPException(status_code=409, detail="用户名已存在")
return crud.update_user(db, user, user_in)
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int, db: SessionDep):
"""删除用户（连带删除其文章）"""
user = crud.get_user(db, user_id)
if user is None:
raise HTTPException(status_code=404, detail="用户不存在")
crud.delete_user(db, user)
观察路由层的模式，每个接口都是同一个节奏：
校验参数（FastAPI 自动） → 查数据 → 业务检查（不存在？重复？）→ 调 crud → 返回
return crud.create_user(...) 返回的是 SQLAlchemy 对象，靠 UserOut 的 from_attributes 自动转 JSON，密码字段被自然过滤掉——第二、三部分的知识在这里完成了会师。
4. 挂载路由：更新 app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app import models # noqa: F401
from app.database import Base, engine
from app.routers import users
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(engine)
yield
app = FastAPI(title="博客 API", version="1.0.0", lifespan=lifespan)
app.include_router(users.router)
@app.get("/")
def read_root():
return {"message": "博客 API 正在运行"}
5. 完整测试流程
启动 fastapi dev app/main.py，打开 /docs，按顺序测试：
- POST /users 创建用户
{"username": "xiaoming", "email": "xm@qq.com", "password": "123456"}→ 应返回 201，响应里有 id、created_at，没有密码 - 再用相同用户名创建一次 → 应返回 409「用户名已存在」
- 密码传
"123"→ 应返回 422（长度校验） - GET /users → 列表里有刚创建的用户
- GET /users/1 → 返回该用户；GET /users/999 → 404
- PATCH /users/1 传
{"username": "daming"}→ 只有用户名变了 - DELETE /users/1 → 204；再 GET → 404
7 步全部符合预期，你就完成了第一个「真正的」后端资源接口！用 DB Browser 打开 blog.db 看看 users 表里的真实数据（注意 hashed_password 列存的是加过工的字符串）。
本章小结
- schemas 定义输入输出；
from_attributes=True让 Pydantic 能读 SQLAlchemy 对象 - crud 层只管数据库操作、不管 HTTP；路由层负责业务检查和错误响应
- 创建前查重复（409）、操作前查存在（404）是资源接口的标准动作
db.refresh()拿回数据库生成的字段
下一章：4.4 文章与标签接口 —— 处理带关联关系的资源。