第 18 章：综合实战 —— 构建完整 REST API

前面 17 章把零件都拆开讲过了。本章把它们装成一台能跑的机器：一个有用户、有认证、有权限、有分页、有测试、有 Docker 的博客 API。读完它，你应该能在半天内从零搭出同等规模的服务。

学习目标

能把第 11 章的分层结构（core / db / models / schemas / repositories / services / api）落到真实项目上
掌握领域异常（Domain Exception）与 HTTP 状态码的解耦，理解为什么 Service 层不该 raise HTTPException
能实现「资源属主或管理员」这类复合权限，并知道依赖参数名与 Path 模板的绑定关系
能设计统一的输入/输出 Schema 与 Page[T] 通用分页响应
能独立完成「注册 → 登录 → 发文 → 越权被拒 → 管理员删除 → 分页」的端到端测试
能写出项目的 Dockerfile 与 docker-compose.yml，并用验收清单自查

📌 本章只给关键片段，不是逐行可复制的完整工程。每个模块取最核心的 20~40 行，其余在文中标注为「留作练习」。

18.1 需求说明：博客 API

模块	能力	角色要求
用户	邮箱注册、登录换 token、查看/修改自己的资料	注册登录匿名；资料需登录
文章	创建、按 id 查看、修改、删除	创建需登录；改删仅作者本人或管理员
文章列表	分页 + 按标签过滤 + 按作者过滤	匿名仅可读已发布
评论	发表、列出某文章评论、删除	发表需登录；删除仅作者本人或管理员
管理	删除任意文章/评论、查看未发布内容	仅 `admin`

明确不做：草稿自动保存、点赞收藏、全文检索、图片上传、邮件验证。范围收敛是项目能收尾的前提。

必须写进代码的业务规则：

slug 全局唯一，冲突返回 409。
tags 规范化：去首尾空白、转小写、去重、最多 5 个。
未发布（published=False）文章仅作者本人与管理员可见，对其他人返回 404 而非 403（不暴露"存在但不可见"）。
删除用户时级联删除其文章与评论。

18.2 技术选型与项目结构

沿用全书基线：FastAPI 0.141.x + Pydantic 2.13.x + SQLAlchemy 2.0（异步）+ Alembic + PostgreSQL 16 + asyncpg；认证用 PyJWT + pwdlib[argon2]；测试用 pytest + pytest-asyncio + httpx。数据库选 PostgreSQL 而不是 SQLite，因为标签用 ARRAY 存储、且依赖可靠的 ON DELETE CASCADE。

text

blog-api/
├── app/
│   ├── main.py                 # create_app()：装配路由、中间件、异常处理器
│   ├── errors.py               # 领域异常 → HTTP 状态码
│   ├── core/
│   │   ├── config.py           # Settings（pydantic-settings）
│   │   └── security.py         # 密码哈希 + JWT 签发校验
│   ├── db/
│   │   ├── base.py             # DeclarativeBase + TimestampMixin
│   │   └── session.py          # engine / async_sessionmaker / get_db
│   ├── models/                 # user.py · post.py · comment.py
│   ├── schemas/                # common.py · user.py · post.py · comment.py
│   ├── repositories/           # post.py · comment.py
│   ├── services/               # exceptions.py · post.py · comment.py
│   └── api/
│       ├── deps.py             # 会话、Service、当前用户等公共依赖
│       └── v1/                 # auth.py · users.py · posts.py · comments.py
├── tests/                      # conftest.py · test_auth.py · test_posts.py
├── migrations/                 # alembic init 生成
├── alembic.ini
├── Dockerfile
├── docker-compose.yml
├── .env.example
└── pyproject.toml

分层调用链

单向依赖：api → services → repositories → models。反向 import 一定意味着分层错了——路由层不该出现 select()，仓储层不该出现 HTTPException。

18.3 配置与数据库

`app/core/config.py`

python

from functools import lru_cache

from pydantic import Field, computed_field
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )

    app_name: str = "Blog API"
    debug: bool = False
    api_v1_prefix: str = "/api/v1"

    secret_key: str = Field(min_length=32, description="JWT 签名密钥")
    jwt_algorithm: str = "HS256"
    access_token_expire_minutes: int = 60

    postgres_user: str = "blog"
    postgres_password: str = "blog"
    postgres_host: str = "localhost"
    postgres_port: int = 5432
    postgres_db: str = "blog"

    cors_origins: list[str] = ["http://localhost:5173"]

    @computed_field
    @property
    def database_url(self) -> str:
        return (
            f"postgresql+asyncpg://{self.postgres_user}:{self.postgres_password}"
            f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
        )


@lru_cache
def get_settings() -> Settings:
    return Settings()

lru_cache 保证环境变量只解析一次；computed_field 让 database_url 能出现在 model_dump() 里（调试方便），同时不允许被环境变量覆盖，避免密码在两处维护。

`app/db/base.py`

python

from datetime import datetime

from sqlalchemy import DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    """Alembic 通过 Base.metadata 比对表结构。"""


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), nullable=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
    )

时间由数据库生成（server_default=func.now()），而不是 Python 侧 default=datetime.utcnow：多实例部署时不会因机器时钟漂移而错乱。

`app/db/session.py`

python

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

from app.core.config import get_settings

settings = get_settings()

engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
)

SessionLocal = async_sessionmaker(
    bind=engine, class_=AsyncSession, expire_on_commit=False, autoflush=False
)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()      # 一个请求 = 一个事务
        except Exception:
            await session.rollback()
            raise

🔑 事务边界放在依赖里最省心：Service 只 flush() 拿主键，提交/回滚交给 get_db。路由里不再出现 await session.commit()，也就不会"忘了提交，数据没落库"。expire_on_commit=False 则保证提交后仍能读属性，序列化响应时不会触发额外 SELECT。

18.4 数据模型

python

# app/models/user.py
class User(Base, TimestampMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(20), default="user", server_default="user")
    is_active: Mapped[bool] = mapped_column(default=True, server_default="true")

    posts: Mapped[list["Post"]] = relationship(
        back_populates="author", cascade="all, delete-orphan", passive_deletes=True
    )


# app/models/post.py
class Post(Base, TimestampMixin):
    __tablename__ = "posts"
    __table_args__ = (
        # 列表页最常见的查询：已发布 + 按时间倒序
        Index("ix_posts_published_created_at", "published", "created_at"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), index=True)
    slug: Mapped[str] = mapped_column(String(220), unique=True, index=True)
    content: Mapped[str] = mapped_column(Text)
    tags: Mapped[list[str]] = mapped_column(ARRAY(String(30)), default=list)
    published: Mapped[bool] = mapped_column(default=False, server_default="false")
    author_id: Mapped[int] = mapped_column(
        ForeignKey("users.id", ondelete="CASCADE"), index=True
    )

    # lazy="selectin"：一次 WHERE id IN (...) 把作者捞出来，避免 N+1
    author: Mapped["User"] = relationship(back_populates="posts", lazy="selectin")
    comments: Mapped[list["Comment"]] = relationship(
        back_populates="post", cascade="all, delete-orphan", passive_deletes=True
    )


# app/models/comment.py
class Comment(Base, TimestampMixin):
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str] = mapped_column(Text)
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id", ondelete="CASCADE"), index=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)

    post: Mapped["Post"] = relationship(back_populates="comments")
    author: Mapped["User"] = relationship(back_populates="comments", lazy="selectin")

（上面的 import 与 TYPE_CHECKING 块留作练习，按第 10 章的写法补全即可。）三个模型共用约定：主键统一 id；外键统一 <表名单数>_id；外键全部建索引，否则级联删除会全表扫；都继承 TimestampMixin。

生成首个迁移：

bash

uv add --dev alembic
uv run alembic init -t async migrations
# migrations/env.py：target_metadata = Base.metadata，并 import 全部模型
uv run alembic revision --autogenerate -m "init users/posts/comments"
uv run alembic upgrade head

⚠️ --autogenerate 靠「已导入到 Base.metadata 的模型」比对，没 import 的模型会被判定为该删表，生成的迁移里带 DROP TABLE。每次新增模型都要在 app/models/__init__.py 里 import 一次。

18.5 Schema 层

`app/schemas/common.py`：通用分页

python

from math import ceil
from typing import Generic, TypeVar

from pydantic import BaseModel, Field, computed_field

T = TypeVar("T")


class Page(BaseModel, Generic[T]):
    """所有列表接口的统一响应形状。"""

    items: list[T]
    total: int = Field(ge=0, description="满足条件的总条数")
    page: int = Field(ge=1)
    size: int = Field(ge=1)

    @computed_field
    @property
    def pages(self) -> int:
        return ceil(self.total / self.size) if self.total else 0

`app/schemas/post.py`

python

from pydantic import BaseModel, ConfigDict, Field, field_validator


class PostCreate(BaseModel):
    """输入模型：只含客户端能决定的东西，绝无 id / author_id。"""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    title: str = Field(min_length=1, max_length=200)
    slug: str = Field(min_length=1, max_length=220, pattern=r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
    content: str = Field(min_length=1)
    tags: list[str] = Field(default_factory=list, max_length=10)
    published: bool = False

    @field_validator("tags", mode="after")
    @classmethod
    def dedupe_tags(cls, v: list[str]) -> list[str]:
        return list(dict.fromkeys(t.strip().lower() for t in v if t.strip()))


class PostUpdate(BaseModel):
    """更新模型：字段全可选，配合 model_dump(exclude_unset=True) 做局部更新。"""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    title: str | None = Field(default=None, min_length=1, max_length=200)
    content: str | None = Field(default=None, min_length=1)
    tags: list[str] | None = Field(default=None, max_length=10)
    published: bool | None = None


class PostOut(BaseModel):
    """输出模型：由 ORM 对象直接构造，敏感字段在结构上就不存在。"""

    model_config = ConfigDict(from_attributes=True)

    id: int
    title: str
    slug: str
    content: str
    tags: list[str]
    published: bool
    author: UserOut
    created_at: datetime
    updated_at: datetime

输入模型 ≠ 输出模型

维度	`PostCreate` / `PostUpdate`	`PostOut`
用途	解析客户端 JSON	序列化 ORM 对象
`from_attributes`	不需要	必须 `True`
`extra`	`"forbid"`（拼错字段立刻 422）	默认 `"ignore"`
含 `id` / `created_at`	不含	含
含 `author_id` / `hashed_password`	不含（不能由客户端指定）	不含

「密码绝不出现」是结构保证，不是运行时检查：UserOut（id / username / email / role / created_at）里根本没有 hashed_password 这个字段，误传 ORM 对象也不会泄露。UserCreate 有 password、UserOut 没有——它们是两个独立的类，不存在"给字段打标记来隐藏"的操作。

18.6 Repository 层

PostRepository 只回答「怎么查、怎么写」，不做业务判断。

python

# app/repositories/post.py
class PostRepository:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def get(self, post_id: int) -> Post | None:
        return await self.session.get(Post, post_id)

    async def get_by_slug(self, slug: str) -> Post | None:
        result = await self.session.execute(select(Post).where(Post.slug == slug))
        return result.scalar_one_or_none()

    async def list_paged(
        self,
        *,
        offset: int,
        limit: int,
        tag: str | None = None,
        author_id: int | None = None,
        published: bool | None = True,
    ) -> tuple[list[Post], int]:
        conditions = []
        if tag is not None:
            conditions.append(Post.tags.any(tag))       # 生成 tag = ANY(posts.tags)
        if author_id is not None:
            conditions.append(Post.author_id == author_id)
        if published is not None:
            conditions.append(Post.published.is_(published))

        base = select(Post).where(*conditions)
        total = await self.session.scalar(select(func.count()).select_from(base.subquery()))
        result = await self.session.execute(
            base.order_by(Post.created_at.desc(), Post.id.desc()).offset(offset).limit(limit)
        )
        return list(result.scalars().all()), total or 0

    async def create(self, data: dict[str, object]) -> Post:
        post = Post(**data)
        self.session.add(post)
        await self.session.flush()          # 拿自增 id，提交交给 get_db
        await self.session.refresh(post)
        return post

    async def update(self, post: Post, data: dict[str, object]) -> Post:
        for key, value in data.items():
            setattr(post, key, value)
        await self.session.flush()
        return post

    async def delete(self, post: Post) -> None:
        await self.session.delete(post)
        await self.session.flush()

排序必须稳定。只按 created_at DESC 排序时，同一秒创建的记录在翻页时可能重复或丢失，所以加 id DESC 兜底。
total 与 items 共享同一套 conditions。用同一个 base 构造 count() 子查询，不会出现"数量对不上列表"。
flush() 不是 commit()。仓储只把 SQL 发出去拿主键，提交时机由上层决定。

18.7 Service 层

领域异常：`app/services/exceptions.py`

python

class DomainError(Exception):
    """领域异常基类。Service 层只抛这一族异常，不 import fastapi。"""


class NotFoundError(DomainError):
    def __init__(self, resource: str, key: object) -> None:
        super().__init__(f"{resource} {key!r} 不存在")
        self.resource = resource
        self.key = key


class PermissionDeniedError(DomainError):
    pass


class ConflictError(DomainError):
    pass


class DomainValidationError(DomainError):
    pass

`app/services/post.py`

python

MAX_PAGE_SIZE = 100
MAX_TAGS = 5
TAG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,29}$")


class PostService:
    def __init__(self, repo: PostRepository) -> None:
        self.repo = repo

    # ---------- 纯函数式规则，可脱离数据库单测 ----------

    @staticmethod
    def normalize_tags(tags: list[str]) -> list[str]:
        normalized: list[str] = []
        for raw in tags:
            tag = raw.strip().lower()
            if not TAG_RE.fullmatch(tag):
                raise DomainValidationError(f"非法标签：{raw!r}，只允许小写字母、数字与连字符")
            if tag not in normalized:
                normalized.append(tag)
        if len(normalized) > MAX_TAGS:
            raise DomainValidationError(f"标签最多 {MAX_TAGS} 个")
        return normalized

    @staticmethod
    def resolve_paging(page: int, size: int) -> tuple[int, int]:
        if page < 1 or not 1 <= size <= MAX_PAGE_SIZE:
            raise DomainValidationError(f"分页参数越界：page={page}, size={size}")
        return (page - 1) * size, size

    # ---------- 业务动作 ----------

    async def get_or_404(self, post_id: int) -> Post:
        post = await self.repo.get(post_id)
        if post is None:
            raise NotFoundError("文章", post_id)
        return post

    async def get_visible_or_404(self, post_id: int, user: User | None) -> Post:
        post = await self.get_or_404(post_id)
        if not post.published:
            is_owner = user is not None and post.author_id == user.id
            is_admin = user is not None and user.role == "admin"
            if not (is_owner or is_admin):
                raise NotFoundError("文章", post_id)      # 不暴露"存在但不可见"
        return post

    async def create(self, data: PostCreate, *, author: User) -> Post:
        if await self.repo.get_by_slug(data.slug) is not None:
            raise ConflictError(f"slug {data.slug!r} 已被占用")
        payload = data.model_dump()
        payload["tags"] = self.normalize_tags(data.tags)
        payload["author_id"] = author.id         # 作者只能是自己，不接受客户端指定
        return await self.repo.create(payload)

    async def update(self, post_id: int, data: PostUpdate, *, user: User) -> Post:
        post = await self.get_or_404(post_id)
        self.ensure_can_modify(post, user)
        payload = data.model_dump(exclude_unset=True)
        if payload.get("tags") is not None:
            payload["tags"] = self.normalize_tags(payload["tags"])
        return await self.repo.update(post, payload)

    async def delete(self, post_id: int, *, user: User) -> None:
        post = await self.get_or_404(post_id)
        self.ensure_can_modify(post, user)
        await self.repo.delete(post)

    async def list_page(
        self,
        *,
        page: int,
        size: int,
        tag: str | None = None,
        author_id: int | None = None,
        published: bool | None = True,
    ) -> Page[PostOut]:
        offset, limit = self.resolve_paging(page, size)
        posts, total = await self.repo.list_paged(
            offset=offset, limit=limit, tag=tag, author_id=author_id, published=published
        )
        return Page[PostOut](
            items=[PostOut.model_validate(p) for p in posts],
            total=total,
            page=page,
            size=size,
        )

    @staticmethod
    def ensure_can_modify(post: Post, user: User) -> None:
        if post.author_id != user.id and user.role != "admin":
            raise PermissionDeniedError("只有作者本人或管理员可以修改/删除该文章")

为什么 Service 不抛 HTTPException：一旦 Service import 了 fastapi，它就绑死在 HTTP 上——后台任务、CLI、gRPC 都复用不了，测试也只能断言状态码而非业务结果。领域异常让 Service 只描述发生了什么，由 errors.py 决定翻译成哪个状态码。

错误示范 vs 正确示范

❌ 不推荐：

python

async def update(self, post_id: int, data: PostUpdate, user: User) -> Post:
    post = await self.repo.get(post_id)
    if post is None:
        raise HTTPException(status_code=404, detail="Not found")
    if post.author_id != user.id:
        raise HTTPException(status_code=403, detail="Forbidden")
    return await self.repo.update(post, data.model_dump())   # 未传字段被 None 覆盖

✅ 推荐：

python

async def update(self, post_id: int, data: PostUpdate, *, user: User) -> Post:
    post = await self.get_or_404(post_id)             # 抛 NotFoundError
    self.ensure_can_modify(post, user)                # 抛 PermissionDeniedError
    return await self.repo.update(post, data.model_dump(exclude_unset=True))

顺带修掉了另一个隐患：不写 exclude_unset=True，客户端没传的字段会被 None 覆盖，PATCH 变成"清空一切"。

18.8 认证接入

复用第 12 章的 get_current_user（从 Authorization: Bearer 解出用户）与 require_role（角色守卫），再补一个"可选用户"变体与复合权限依赖。

python

# app/api/deps.py
# auto_error=False：缺 token 时返回 None 而不是直接 401，匿名接口才能复用
oauth2_scheme_optional = OAuth2PasswordBearer(
    tokenUrl=f"{settings.api_v1_prefix}/auth/token", auto_error=False
)

SessionDep = Annotated[AsyncSession, Depends(get_db)]


async def get_current_user(
    session: SessionDep, token: Annotated[str, Depends(oauth2_scheme)]
) -> User:
    payload = decode_access_token(token)        # 校验签名与 exp，失败抛 401
    user = await session.get(User, int(payload["sub"]))
    if user is None or not user.is_active:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "用户不存在或已停用")
    return user


async def get_current_user_optional(
    session: SessionDep, token: Annotated[str | None, Depends(oauth2_scheme_optional)] = None
) -> User | None:
    """匿名可访问的接口用它：有 token 就解出来，没有或无效就返回 None。"""
    if token is None:
        return None
    try:
        return await get_current_user(session, token)
    except HTTPException:
        return None


CurrentUser = Annotated[User, Depends(get_current_user)]
OptionalUser = Annotated[User | None, Depends(get_current_user_optional)]


def require_role(*roles: str):
    async def checker(user: CurrentUser) -> User:
        if user.role not in roles:
            raise HTTPException(status.HTTP_403_FORBIDDEN, "权限不足")
        return user

    return checker


AdminUser = Annotated[User, Depends(require_role("admin"))]


def get_post_service(session: SessionDep) -> PostService:
    return PostService(PostRepository(session))


PostServiceDep = Annotated[PostService, Depends(get_post_service)]

「作者本人或管理员」有两种落点：依赖层（Depends(get_owned_post) 直接返回已鉴权的 Post，适合规则稳定、多端点复用的场景）与 Service 层（ensure_can_modify，适合规则会随业务演进，比如以后加"协作者也能改"）。本项目两者都留：依赖负责快速拦截，Service 负责最终判定。

python

# app/api/v1/_permissions.py
async def get_owned_post(
    post_id: Annotated[int, Path(ge=1)],
    service: PostServiceDep,
    user: CurrentUser,
) -> Post:
    post = await service.get_or_404(post_id)
    if post.author_id != user.id and user.role != "admin":
        raise PermissionDeniedError("只有作者本人或管理员可以修改/删除该文章")
    return post


OwnedPost = Annotated[Post, Depends(get_owned_post)]

🔑 依赖参数的坑：get_owned_post 的参数名 post_id 必须与路由模板 {post_id} 逐字一致，否则 FastAPI 会把它当成查询参数并返回 422 field required。同一个请求内 Depends(get_db) 会被缓存复用，所以 get_owned_post 与 get_current_user 拿到的是同一个 AsyncSession——这也是能安全传递 ORM 对象的前提。

18.9 路由层

`app/api/v1/posts.py`

python

router = APIRouter(prefix="/posts", tags=["posts"])


@router.get("", response_model=Page[PostOut], summary="分页查询文章列表")
async def list_posts(
    service: PostServiceDep,
    user: OptionalUser,
    page: Annotated[int, Query(ge=1, description="页码，从 1 开始")] = 1,
    size: Annotated[int, Query(ge=1, le=100, description="每页条数")] = 20,
    tag: Annotated[str | None, Query(max_length=30)] = None,
    author_id: Annotated[int | None, Query(ge=1)] = None,
) -> Page[PostOut]:
    # 匿名只看已发布；登录用户看自己的列表时可含未发布
    published = True if user is None or author_id != user.id else None
    return await service.list_page(
        page=page, size=size, tag=tag, author_id=author_id, published=published
    )


@router.post("", response_model=PostOut, status_code=status.HTTP_201_CREATED, summary="创建文章")
async def create_post(
    payload: PostCreate, service: PostServiceDep, user: CurrentUser
) -> PostOut:
    return PostOut.model_validate(await service.create(payload, author=user))


@router.get("/{post_id}", response_model=PostOut, summary="查看文章详情")
async def get_post(
    post_id: Annotated[int, Path(ge=1)], service: PostServiceDep, user: OptionalUser
) -> PostOut:
    return PostOut.model_validate(await service.get_visible_or_404(post_id, user))


@router.patch("/{post_id}", response_model=PostOut, summary="局部更新文章")
async def update_post(
    post: OwnedPost, payload: PostUpdate, service: PostServiceDep, user: CurrentUser
) -> PostOut:
    # 依赖已确认权限；Service 内部再判一次，防止未来规则分叉时漏改
    return PostOut.model_validate(await service.update(post.id, payload, user=user))


@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT, summary="删除文章")
async def delete_post(
    post: OwnedPost, service: PostServiceDep, user: CurrentUser
) -> None:
    await service.delete(post.id, user=user)

`app/api/v1/comments.py`

python

router = APIRouter(prefix="/posts/{post_id}/comments", tags=["comments"])


@router.get("", response_model=list[CommentOut], summary="列出文章评论")
async def list_comments(
    post_id: Annotated[int, Path(ge=1)], service: CommentServiceDep
) -> list[CommentOut]:
    return [CommentOut.model_validate(c) for c in await service.list_by_post(post_id)]


@router.post("", response_model=CommentOut, status_code=status.HTTP_201_CREATED, summary="发表评论")
async def create_comment(
    post_id: Annotated[int, Path(ge=1)],
    payload: CommentCreate,
    service: CommentServiceDep,
    user: CurrentUser,
) -> CommentOut:
    return CommentOut.model_validate(await service.create(post_id, payload, author=user))

`app/main.py`：组装

python

from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await engine.dispose()          # 关闭时释放连接池，进程能干净退出


def create_app() -> FastAPI:
    settings = get_settings()
    app = FastAPI(title=settings.app_name, version="1.0.0", lifespan=lifespan)

    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins,     # 精确列举，不用 ["*"] + credentials
        allow_credentials=True,
        allow_methods=["GET", "POST", "PATCH", "DELETE", "OPTIONS"],
        allow_headers=["Authorization", "Content-Type"],
    )

    register_exception_handlers(app)

    prefix = settings.api_v1_prefix
    app.include_router(auth.router, prefix=prefix)
    app.include_router(users.router, prefix=prefix)
    app.include_router(posts.router, prefix=prefix)
    app.include_router(comments.router, prefix=prefix)
    return app


app = create_app()

用 create_app() 工厂而不是模块级裸 app：测试能造多个互不干扰的实例，也方便按环境换配置——第 16 章的 dependency_overrides 依赖这个结构。

18.10 异常与统一响应

app/errors.py 是整个项目里唯一知道状态码的地方（除了路由上的 status_code=）。

python

from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(NotFoundError)
    async def not_found(request: Request, exc: NotFoundError) -> JSONResponse:
        return JSONResponse(
            status_code=status.HTTP_404_NOT_FOUND,
            content={"detail": str(exc), "code": "not_found"},
        )

    @app.exception_handler(PermissionDeniedError)
    async def denied(request: Request, exc: PermissionDeniedError) -> JSONResponse:
        return JSONResponse(
            status_code=status.HTTP_403_FORBIDDEN,
            content={"detail": str(exc), "code": "forbidden"},
        )

    @app.exception_handler(ConflictError)
    async def conflict(request: Request, exc: ConflictError) -> JSONResponse:
        return JSONResponse(
            status_code=status.HTTP_409_CONFLICT,
            content={"detail": str(exc), "code": "conflict"},
        )

    @app.exception_handler(DomainValidationError)
    async def invalid(request: Request, exc: DomainValidationError) -> JSONResponse:
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content={"detail": str(exc), "code": "validation_error"},
        )

    @app.exception_handler(RequestValidationError)
    async def schema_invalid(request: Request, exc: RequestValidationError) -> JSONResponse:
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content={"detail": exc.errors(), "code": "request_validation_error"},
        )

领域异常	触发场景	HTTP	`code`
`NotFoundError`	id 不存在；未发布文章对无权者不可见	404	`not_found`
`PermissionDeniedError`	非作者、非管理员改删资源	403	`forbidden`
`ConflictError`	`slug` 或邮箱重复	409	`conflict`
`DomainValidationError`	标签非法、分页越界	422	`validation_error`
`RequestValidationError`	Pydantic Schema 校验失败（框架抛出）	422	`request_validation_error`

⚠️ 不要给裸 Exception 注册返回 500 JSON 的处理器。开发时它会吞掉堆栈，让调试变成猜谜。真要统一 500 响应，处理器里先 logger.exception(...) 再返回。

18.11 测试

测试库用 PostgreSQL 而不是 SQLite：tags 是 ARRAY 类型 SQLite 不认，ON DELETE CASCADE 行为也有差异。测试数据库与生产同款是最省事的原则。

python

# tests/conftest.py
TEST_DATABASE_URL = os.getenv(
    "TEST_DATABASE_URL", "postgresql+asyncpg://blog:blog@localhost:5432/blog_test"
)


@pytest_asyncio.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
    engine = create_async_engine(TEST_DATABASE_URL, poolclass=NullPool)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

    async def override_get_db() -> AsyncGenerator[AsyncSession, None]:
        async with maker() as session:
            try:
                yield session
                await session.commit()
            except Exception:
                await session.rollback()
                raise

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()
    await engine.dispose()

一条用例走完全流程：

python

# tests/test_posts.py
pytestmark = pytest.mark.asyncio


async def register_and_login(client, name: str) -> dict[str, str]:
    await client.post(
        "/api/v1/auth/register",
        json={"username": name, "email": f"{name}@example.com", "password": "s3cret-pass"},
    )
    resp = await client.post(
        "/api/v1/auth/token", data={"username": name, "password": "s3cret-pass"}
    )
    return {"Authorization": f"Bearer {resp.json()['access_token']}"}


async def test_full_post_lifecycle(client, admin_headers):
    alice = await register_and_login(client, "alice")
    bob = await register_and_login(client, "bob")

    # 1) 创建：201，标签被规范化，响应不含密码
    created = await client.post(
        "/api/v1/posts",
        headers=alice,
        json={
            "title": "FastAPI 分层实践",
            "slug": "fastapi-layering",
            "content": "正文……",
            "tags": ["FastAPI", " fastapi ", "Architecture"],
            "published": True,
        },
    )
    assert created.status_code == 201, created.text
    body = created.json()
    assert body["tags"] == ["fastapi", "architecture"]        # 小写 + 去重
    assert "hashed_password" not in body["author"]
    post_id = body["id"]

    # 2) slug 冲突 → 409
    dup = await client.post(
        "/api/v1/posts", headers=bob,
        json={"title": "另一篇", "slug": "fastapi-layering", "content": "正文……"},
    )
    assert dup.status_code == 409

    # 3) 他人修改 → 403
    forbidden = await client.patch(
        f"/api/v1/posts/{post_id}", headers=bob, json={"title": "被篡改"}
    )
    assert forbidden.status_code == 403

    # 4) 作者本人修改 → 200，且未传字段（content）保持不变
    updated = await client.patch(
        f"/api/v1/posts/{post_id}", headers=alice, json={"title": "新标题"}
    )
    assert updated.status_code == 200
    assert updated.json()["content"] == "正文……"

    # 5) 匿名分页 → 200，Page 结构完整
    page = await client.get("/api/v1/posts", params={"page": 1, "size": 10})
    assert set(page.json()) == {"items", "total", "page", "size", "pages"}

    # 6) 管理员删任意文章 → 204；再取 → 404
    assert (
        await client.delete(f"/api/v1/posts/{post_id}", headers=admin_headers)
    ).status_code == 204
    assert (await client.get(f"/api/v1/posts/{post_id}")).status_code == 404

admin_headers fixture 留作练习。建议再补四类用例：无 token 访问受保护端点（401）、page=0 或 size=999（422）、未发布文章对他人 404 对作者 200、分页第二页无重复 id。

bash

uv run pytest -q --cov=app --cov-report=term-missing

18.12 Docker 化

dockerfile

FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 UV_PROJECT_ENVIRONMENT=/usr/local

COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
WORKDIR /app

# 先复制依赖清单：只改代码时不会重装依赖，利用镜像层缓存
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

COPY . .
EXPOSE 8000

CMD ["uv", "run", "fastapi", "run", "app/main.py", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

yaml

services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: blog
      POSTGRES_PASSWORD: blog
      POSTGRES_DB: blog
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U blog -d blog"]
      interval: 5s
      timeout: 3s
      retries: 10
    ports:
      - "5432:5432"

  migrate:
    build: .
    command: ["uv", "run", "alembic", "upgrade", "head"]
    environment:
      POSTGRES_HOST: db
      SECRET_KEY: ${SECRET_KEY}
    depends_on:
      db:
        condition: service_healthy
    restart: "no"

  api:
    build: .
    environment:
      POSTGRES_HOST: db
      SECRET_KEY: ${SECRET_KEY}
      DEBUG: "false"
    depends_on:
      migrate:
        condition: service_completed_successfully
    ports:
      - "8000:8000"

volumes:
  pgdata:

migrate 做成一次性服务、api 用 service_completed_successfully 等它跑完，启动顺序就是确定的：库健康 → 迁移完成 → API 起来。这比在 lifespan 里偷偷跑 create_all() 可靠得多。

ini

# .env.example
APP_NAME=Blog API
DEBUG=true
SECRET_KEY=change-me-to-a-random-32-bytes-string
ACCESS_TOKEN_EXPIRE_MINUTES=60

POSTGRES_USER=blog
POSTGRES_PASSWORD=blog
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=blog

CORS_ORIGINS=["http://localhost:5173"]

bash

cp .env.example .env
uv run alembic upgrade head
uv run fastapi dev app/main.py

18.13 验收清单

做完这些，才算项目达标。建议直接抄进项目 README.md：

#	检查项	达标标准
1	目录分层	存在 `core` / `db` / `models` / `schemas` / `repositories` / `services` / `api/v1`，无反向依赖
2	配置外置	无硬编码连接串与密钥；`.env.example` 齐全；`Settings` 用 `get_settings()` 缓存
3	迁移可用	`alembic upgrade head` 从空库建出全部表；`downgrade base` 能回滚
4	密码安全	库中只存 Argon2 哈希；任何响应体都不含 `password` / `hashed_password`
5	认证	`/auth/register` 返回 201；`/auth/token` 用 `OAuth2PasswordRequestForm` 换取 JWT
6	鉴权	无 token 访问受保护端点 401；越权改删 403
7	管理员	`admin` 可删任意文章与评论，非管理员不可
8	资源语义	创建 201、删除 204、其余 200；列表接口统一返回 `Page[T]`
9	分页	`page` / `size` 有边界校验；`total` 与过滤条件一致；排序稳定（`created_at DESC, id DESC`）
10	过滤	`?tag=` 与 `?author_id=` 生效；标签经小写去重规范化
11	可见性	未发布文章对匿名/他人返回 404，对作者与管理员可见
12	更新语义	PATCH 用 `exclude_unset=True`，未传字段不被置空
13	错误响应	全部错误形如 `{"detail": ..., "code": ...}`；422 保留字段路径
14	异常解耦	`services/` 与 `repositories/` 中不出现 `import fastapi`
15	无 N+1	文章列表的 SQL 条数不随记录数增长（`lazy="selectin"`）
16	测试	端到端用例覆盖注册/登录/发文/越权/删除/分页；`pytest --cov` 行覆盖 ≥ 80%
17	并发安全	并发创建同名 slug：一个成功、一个 409（唯一索引兜底）
18	Docker	`docker compose up` 一条命令起全栈，迁移自动执行
19	文档	`/docs` 可交互；每个路由都有 `summary` 与 `response_model`
20	关闭干净	进程收到 SIGTERM 后释放连接池，无 "Event loop is closed" 报错

常见坑与排查

现象	原因	解决
`autogenerate` 生成一堆 `DROP TABLE`	模型没被 import，`Base.metadata` 里没有它们	在 `app/models/__init__.py` 或 `migrations/env.py` 里显式 import 全部模型
`MissingGreenlet: greenlet_spawn has not been called`	异步上下文里触发了同步惰性加载（如提交后访问关系）	关系设 `lazy="selectin"` 或查询时 `.options(selectinload(...))`；`expire_on_commit=False`
列表返回的 `author` 为空，或伴随几十条 SELECT	关系默认 `lazy="select"`	见上一条；用 `selectinload` 预加载
依赖里写了 `post_id` 却收到 422 `field required`	参数名与路径模板 `{post_id}` 不一致，被当成查询参数	依赖参数名必须与路径模板变量名逐字一致
PATCH 后未传字段全变 `null`	用了 `model_dump()` 而不是 `model_dump(exclude_unset=True)`	局部更新一律 `exclude_unset=True`
翻页时第 1、2 页出现同一条记录	排序键不唯一，同秒创建的记录顺序不定	排序加主键兜底：`order_by(created_at.desc(), id.desc())`
列表 `total` 与 `items` 数量对不上	总数查询漏了某个过滤条件	用同一个 `base` 语句构造 `count()` 子查询
`docker compose up` 时 API 报 relation 不存在	`depends_on` 只保证容器启动，不保证迁移跑完	`migrate` 用 `service_completed_successfully`，`db` 加 `healthcheck`
CORS 报错，配置里明明写了域名	`allow_origins=["*"]` 与 `allow_credentials=True` 不能共存	精确列举来源，本地开发加 `http://localhost:5173`
测试里改的数据下个用例还能看到	没清表，或 `dependency_overrides` 没清	fixture 里 `drop_all` + `create_all`，`yield` 后 `clear()`

本章小结

要点	说明
先定边界	需求表 + "明确不做"清单，是项目能收尾的前提
单向依赖	`api → services → repositories → models`，反向 import 即分层错误
事务边界	放 `get_db` 依赖：请求成功 `commit`、异常 `rollback`，Service 只 `flush`
输入输出分离	`XxxCreate`（`extra="forbid"`）与 `XxxOut`（`from_attributes=True`）是两个类，敏感字段用"不存在"来保证
领域异常	Service 抛 `NotFoundError` / `PermissionDeniedError` / `ConflictError`，`errors.py` 统一映射 404/403/409
复合权限	依赖层 `get_owned_post` 快速拦截，Service `ensure_can_modify` 做最终判定
分页三件套	边界校验 + 同源 `count()` + 稳定排序键
测试库同款	用 PostgreSQL 测，别为了 `ARRAY` 与级联行为去迁就 SQLite
启动顺序	`db` 健康 → `migrate` 完成 → `api` 启动，全部写进 compose

练习题

给文章列表加 ?q= 关键词搜索（title ILIKE %q%），要求 q 长度 2~50、搜索结果同样带 total；说明为什么 title 上的普通 B-tree 索引对这个查询帮助有限，以及可以怎么改进。
实现评论的完整 CRUD，并加一条规则：文章作者可以删除自己文章下的任意评论。分别用"依赖层"和"Service 层"实现，比较两者的可测试性。
把 tags 从 ARRAY(String) 改造成独立的 tags 表 + post_tags 关联表，写出模型与首个 Alembic 迁移，并说明两种方案在"按标签统计文章数"这个查询上的写法差异。
给项目加软删除（deleted_at）：所有查询默认过滤已删除记录，管理员可以"恢复"文章。要求不破坏现有测试，并说明软删除为什么会让 unique(slug) 变麻烦、有哪几种处理方式。
用 hey 或 locust 压测 GET /api/v1/posts?page=1&size=20，找出瓶颈后从"加索引 / 加缓存 / 调连接池"三个方向各试一次，记录 QPS 变化并解释原因。

下一章预告

项目能跑了，接下来把这些经验固化成规范：命名、状态码、性能清单、安全清单，以及一份随时能翻的速查表。
👉 第 19 章：最佳实践与速查表

第 18 章：综合实战 —— 构建完整 REST API ​

学习目标 ​

18.1 需求说明：博客 API ​

18.2 技术选型与项目结构 ​

分层调用链 ​

18.3 配置与数据库 ​

app/core/config.py ​

app/db/base.py ​

app/db/session.py ​

18.4 数据模型 ​

18.5 Schema 层 ​

app/schemas/common.py：通用分页 ​

app/schemas/post.py ​

输入模型 ≠ 输出模型 ​

18.6 Repository 层 ​

18.7 Service 层 ​

领域异常：app/services/exceptions.py ​

app/services/post.py ​

错误示范 vs 正确示范 ​

18.8 认证接入 ​

18.9 路由层 ​

app/api/v1/posts.py ​

app/api/v1/comments.py ​

app/main.py：组装 ​

18.10 异常与统一响应 ​

18.11 测试 ​

18.12 Docker 化 ​

18.13 验收清单 ​

常见坑与排查 ​

本章小结 ​

练习题 ​

下一章预告 ​

第 18 章：综合实战 —— 构建完整 REST API

学习目标

18.1 需求说明：博客 API

18.2 技术选型与项目结构

分层调用链

18.3 配置与数据库

`app/core/config.py`

`app/db/base.py`

`app/db/session.py`

18.4 数据模型

18.5 Schema 层

`app/schemas/common.py`：通用分页

`app/schemas/post.py`

输入模型 ≠ 输出模型

18.6 Repository 层

18.7 Service 层

领域异常：`app/services/exceptions.py`

`app/services/post.py`

错误示范 vs 正确示范

18.8 认证接入

18.9 路由层

`app/api/v1/posts.py`

`app/api/v1/comments.py`

`app/main.py`：组装

18.10 异常与统一响应

18.11 测试

18.12 Docker 化

18.13 验收清单

常见坑与排查

本章小结

练习题

下一章预告