04 文章与标签接口
本章目标：实现文章（Post）的 CRUD，处理「文章-作者」一对多和「文章-标签」多对多关系，学会返回嵌套 JSON。
1. schemas.py 追加文章和标签模型
在 app/schemas.py 末尾追加：
# ---------- 标签 ----------
class TagOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
# ---------- 文章 ----------
class PostCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
content: str = Field(min_length=1)
published: bool = False
author_id: int # 第五部分学了登录后，会改成"取当前登录用户"
tags: list[str] = [] # 创建时直接传标签名列表，如 ["python", "fastapi"]
class PostUpdate(BaseModel):
title: str | None = Field(default=None, min_length=1, max_length=200)
content: str | None = Field(default=None, min_length=1)
published: bool | None = None
tags: list[str] | None = None
class PostOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
title: str
content: str
published: bool
created_at: datetime
updated_at: datetime
author: UserOut # 嵌套返回作者信息！
tags: list[TagOut] = [] # 嵌套返回标签列表！
嵌套响应模型
PostOut 里的 author: UserOut、tags: list[TagOut] 是本章的精华：由于模型上定义了 relationship，返回 Post 对象时 Pydantic 会顺着 post.author、post.tags 属性把关联对象也序列化出来，响应长这样：
{
"id": 1,
"title": "FastAPI 入门",
"content": "...",
"published": true,
"created_at": "2026-07-26T10:00:00",
"updated_at": "2026-07-26T10:00:00",
"author": {"id": 1, "username": "xiaoming", "email": "xm@qq.com", "is_active": true, "created_at": "..."},
"tags": [{"id": 1, "name": "python"}, {"id": 2, "name": "fastapi"}]
}
前端一个请求拿到全部信息，不用再发两次请求查作者和标签。
注意不要循环嵌套：如果
UserOut里再放posts: list[PostOut]，而PostOut里有author: UserOut，就会无限递归。规则：嵌套只往「一个方向」放。
2. crud.py 追加文章和标签操作
在 app/crud.py 末尾追加：
from sqlalchemy.orm import selectinload
# ---------- 标签 ----------
def get_or_create_tags(db: Session, names: list[str]) -> list[models.Tag]:
"""根据名字列表拿标签：已存在的直接用，不存在的创建"""
tags: list[models.Tag] = []
for name in names:
name = name.strip().lower()
if not name:
continue
tag = db.scalars(
select(models.Tag).where(models.Tag.name == name)
).first()
if tag is None:
tag = models.Tag(name=name)
db.add(tag)
tags.append(tag)
return tags
# ---------- 文章 ----------
def get_post(db: Session, post_id: int) -> models.Post | None:
stmt = (
select(models.Post)
.where(models.Post.id == post_id)
.options(
selectinload(models.Post.author), # 预加载作者
selectinload(models.Post.tags), # 预加载标签
)
)
return db.scalars(stmt).first()
def list_posts(db: Session, offset: int = 0, limit: int = 10) -> list[models.Post]:
stmt = (
select(models.Post)
.options(selectinload(models.Post.author), selectinload(models.Post.tags))
.order_by(models.Post.id.desc()) # 新文章在前
.offset(offset)
.limit(limit)
)
return list(db.scalars(stmt).all())
def create_post(db: Session, post_in: schemas.PostCreate) -> models.Post:
post = models.Post(
title=post_in.title,
content=post_in.content,
published=post_in.published,
author_id=post_in.author_id,
tags=get_or_create_tags(db, post_in.tags),
)
db.add(post)
db.commit()
return get_post(db, post.id) # 重新查一遍，带上预加载的关联
def update_post(
db: Session, post: models.Post, post_in: schemas.PostUpdate
) -> models.Post:
data = post_in.model_dump(exclude_unset=True)
tags = data.pop("tags", None) # tags 单独处理
for field, value in data.items():
setattr(post, field, value)
if tags is not None:
post.tags = get_or_create_tags(db, tags) # 整体替换标签
db.commit()
return get_post(db, post.id)
def delete_post(db: Session, post: models.Post) -> None:
db.delete(post)
db.commit()
要点：
- selectinload 预加载：因为
PostOut要序列化 author 和 tags，如果不预加载，每篇文章序列化时都会触发懒加载查询（N+1 问题，见第三部分第 5 章）。列表接口必须预加载 - get_or_create 模式：用户传标签名，存在就复用、不存在就建，非常常用的模式
- 更新时
post.tags = 新列表即可整体替换多对多关联，中间表由 SQLAlchemy 自动增删
3. app/routers/posts.py
from fastapi import APIRouter, HTTPException, status
from app import crud, schemas
from app.database import SessionDep
router = APIRouter(prefix="/posts", tags=["文章"])
@router.post("", response_model=schemas.PostOut, status_code=status.HTTP_201_CREATED)
def create_post(post_in: schemas.PostCreate, db: SessionDep):
"""发布文章"""
if crud.get_user(db, post_in.author_id) is None:
raise HTTPException(status_code=404, detail="作者不存在")
return crud.create_post(db, post_in)
@router.get("", response_model=list[schemas.PostOut])
def list_posts(db: SessionDep, page: int = 1, size: int = 10):
"""文章列表"""
size = min(size, 100)
return crud.list_posts(db, offset=(page - 1) * size, limit=size)
@router.get("/{post_id}", response_model=schemas.PostOut)
def get_post(post_id: int, db: SessionDep):
"""文章详情"""
post = crud.get_post(db, post_id)
if post is None:
raise HTTPException(status_code=404, detail="文章不存在")
return post
@router.patch("/{post_id}", response_model=schemas.PostOut)
def update_post(post_id: int, post_in: schemas.PostUpdate, db: SessionDep):
"""修改文章"""
post = crud.get_post(db, post_id)
if post is None:
raise HTTPException(status_code=404, detail="文章不存在")
return crud.update_post(db, post, post_in)
@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int, db: SessionDep):
"""删除文章"""
post = crud.get_post(db, post_id)
if post is None:
raise HTTPException(status_code=404, detail="文章不存在")
crud.delete_post(db, post)
4. 挂载路由
app/main.py 中：
from app.routers import posts, users
app.include_router(users.router)
app.include_router(posts.router)
5. 追加一个「查某用户的所有文章」接口
在 routers/users.py 里加（体现 REST 风格的资源嵌套路径）：
from app import models
from sqlalchemy import select
from sqlalchemy.orm import selectinload
@router.get("/{user_id}/posts", response_model=list[schemas.PostOut])
def list_user_posts(user_id: int, db: SessionDep):
"""某用户发布的所有文章"""
if crud.get_user(db, user_id) is None:
raise HTTPException(status_code=404, detail="用户不存在")
stmt = (
select(models.Post)
.where(models.Post.author_id == user_id)
.options(selectinload(models.Post.author), selectinload(models.Post.tags))
.order_by(models.Post.id.desc())
)
return list(db.scalars(stmt).all())
6. 测试流程
在 /docs 里：
- 先创建一个用户（拿到 id，比如 1）
- POST /posts：
{"title": "FastAPI 入门", "content": "正文...", "published": true, "author_id": 1, "tags": ["python", "fastapi"]}→ 201，响应里有嵌套的 author 和 tags - 再发一篇带
"tags": ["python"]的文章 → 查看数据库 tags 表，python标签只有一条（复用了） - GET /posts → 列表，每篇都带作者和标签
- PATCH /posts/1：
{"tags": ["web"]}→ 标签被整体替换 - GET /users/1/posts → 该用户的文章
- 删除用户 1，再 GET /posts → 他的文章也没了（级联删除生效）
本章小结
- 响应模型可以嵌套（author: UserOut、tags: list[TagOut]），前提是查询时用
selectinload预加载关联 - 嵌套只往一个方向放，避免循环引用
- get_or_create 处理「传名字建关联」场景；
post.tags = 新列表整体替换多对多 - 资源嵌套路径
/users/{id}/posts表达从属关系
下一章：05-分页搜索排序 —— 把列表接口做专业。