FastAPI 实战开发内容管理系统：从零搭建完整 CMS 后台

前面我们已经学习了 FastAPI 的基础 API 开发、MySQL 数据库操作、JWT 用户认证，以及分页、搜索和筛选等功能。

当这些基础模块逐渐完善后，就可以尝试开发一个更完整的项目：内容管理系统（CMS）。

CMS 是 Content Management System 的缩写，主要用于管理网站文章、分类、标签、用户和内容发布状态。无论是技术博客、新闻网站，还是古诗词内容平台，都可以使用 CMS 统一管理网站数据。

本文将使用 FastAPI、MySQL、SQLAlchemy 2.0 和 JWT，搭建一个具备实际扩展能力的 CMS 后端。我们会从项目结构开始，逐步实现用户认证、文章管理、分类管理、分页搜索、权限控制和生产环境部署思路。

一、项目功能规划

本次开发的 CMS 将围绕内容管理展开。

为了控制教程复杂度，本文先实现一个可运行的基础版本。评论系统、媒体上传、操作审计、多级角色和前端管理界面可以在后续章节中逐步增加。

二、项目技术栈

我们选择以下技术：

• FastAPI：提供 REST API。

• MySQL：存储用户、文章和分类。

• SQLAlchemy 2.0：操作数据库。

• Pydantic：验证请求数据。

• PyJWT：签发和验证登录令牌。

• pwdlib + Argon2：安全地哈希用户密码。

• Uvicorn：运行 FastAPI。

安装依赖：

pip install fastapi "uvicorn[standard]" sqlalchemy pymysql PyJWT "pwdlib[argon2]" python-multipart email-validator

本文采用同步 SQLAlchemy。对于个人网站和中小型 CMS，这种写法容易理解和维护；如果未来需要异步数据库访问，可以再统一迁移到 AsyncSession。

三、创建项目目录

建议从一开始就采用模块化结构，而不是将所有业务代码都放在 main.py 中。

fastapi-cms/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   ├── schemas.py
│   ├── security.py
│   ├── deps.py
│   └── routers/
│       ├── __init__.py
│       ├── auth.py
│       ├── articles.py
│       └── categories.py
├── requirements.txt
└── .env.example

各文件职责如下：

• database.py：数据库连接和会话。

• models.py：数据库模型。

• schemas.py：请求与响应模型。

• security.py：密码哈希和 JWT。

• deps.py：当前用户、管理员等依赖。

• routers/：按业务模块拆分接口。

• main.py：创建 FastAPI 应用并注册路由。

这样的结构便于后续增加标签、用户、媒体资源和系统设置模块。

四、创建 MySQL 数据库

登录 MySQL，创建数据库：

CREATE DATABASE fastapi_cms
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;

建议创建专用数据库账号，不要让 Web 应用直接使用 MySQL root 账号。

CREATE USER 'cms_user'@'localhost'
IDENTIFIED BY '替换为强数据库密码';

GRANT SELECT, INSERT, UPDATE, DELETE
ON fastapi_cms.*
TO 'cms_user'@'localhost';

如果数据库与应用位于不同服务器，应根据实际部署架构调整账号来源限制和网络访问规则。

五、配置数据库连接

创建 app/database.py：

import os

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

DATABASE_URL = os.environ["DATABASE_URL"]

engine = create_engine(
    DATABASE_URL,
    pool_pre_ping=True,
    pool_recycle=1800,
)

SessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    expire_on_commit=False,
)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

开发环境中设置数据库连接字符串：

export DATABASE_URL='mysql+pymysql://cms_user:替换为强数据库密码@127.0.0.1:3306/fastapi_cms?charset=utf8mb4'

生产环境应使用环境变量或安全的配置管理机制保存密码，不要将包含真实凭据的配置文件提交到公开代码仓库。

六、设计数据库模型

本次先创建三张核心数据表：

1. users：管理员及用户信息。

2. categories：内容分类。

3. articles：文章内容。

在 app/models.py 中编写：

from datetime import datetime

from sqlalchemy import (
    Boolean,
    DateTime,
    ForeignKey,
    Integer,
    String,
    Text,
    UniqueConstraint,
    func,
)
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
    )
    username: Mapped[str] = mapped_column(
        String(50),
        unique=True,
        nullable=False,
    )
    hashed_password: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
    )
    is_active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        nullable=False,
    )
    is_admin: Mapped[bool] = mapped_column(
        Boolean,
        default=False,
        nullable=False,
    )


class Category(Base):
    __tablename__ = "categories"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
    )
    name: Mapped[str] = mapped_column(
        String(100),
        unique=True,
        nullable=False,
    )
    slug: Mapped[str] = mapped_column(
        String(120),
        unique=True,
        nullable=False,
    )


class Article(Base):
    __tablename__ = "articles"

    id: Mapped[int] = mapped_column(
        Integer,
        primary_key=True,
    )
    title: Mapped[str] = mapped_column(
        String(200),
        nullable=False,
    )
    slug: Mapped[str] = mapped_column(
        String(220),
        unique=True,
        nullable=False,
    )
    content: Mapped[str] = mapped_column(
        Text,
        nullable=False,
    )
    status: Mapped[str] = mapped_column(
        String(20),
        default="draft",
        nullable=False,
        index=True,
    )
    category_id: Mapped[int | None] = mapped_column(
        ForeignKey("categories.id", ondelete="SET NULL"),
        nullable=True,
        index=True,
    )
    author_id: Mapped[int] = mapped_column(
        ForeignKey("users.id"),
        nullable=False,
        index=True,
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=func.now(),
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime,
        nullable=False,
        server_default=func.now(),
        onupdate=func.now(),
    )

为什么文章需要 slug？

slug 是适合放在 URL 中的内容标识，例如：

/articles/fastapi-cms-guide

它可以使 URL 更容易阅读，也有利于建立稳定的内容链接。

对于中文古诗词网站，可以根据业务需要使用拼音、英文标识或固定 ID 生成 URL。必须保证标识唯一，并且在发布后谨慎修改，避免已有链接失效。

注意：本文中的 slug 唯一约束是数据库层面的保障。创建文章时仍需要捕获唯一约束冲突，避免两个并发请求生成相同标识。

七、定义请求与响应模型

创建 app/schemas.py：

from datetime import datetime

from pydantic import BaseModel, ConfigDict, Field


class LoginRequest(BaseModel):
    username: str = Field(min_length=1, max_length=50)
    password: str = Field(min_length=1, max_length=128)


class TokenResponse(BaseModel):
    access_token: str
    token_type: str = "bearer"


class CategoryCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    slug: str = Field(min_length=1, max_length=120)


class CategoryResponse(BaseModel):
    id: int
    name: str
    slug: str

    model_config = ConfigDict(from_attributes=True)


class ArticleCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    slug: str = Field(min_length=1, max_length=220)
    content: str
    category_id: int | None = None
    status: str = "draft"


class ArticleUpdate(BaseModel):
    title: str | None = Field(None, min_length=1, max_length=200)
    slug: str | None = Field(None, min_length=1, max_length=220)
    content: str | None = None
    category_id: int | None = None
    status: str | None = None


class ArticleResponse(BaseModel):
    id: int
    title: str
    slug: str
    content: str
    status: str
    category_id: int | None
    author_id: int
    created_at: datetime
    updated_at: datetime

    model_config = ConfigDict(from_attributes=True)


class ArticlePage(BaseModel):
    total: int
    page: int
    page_size: int
    items: list[ArticleResponse]

请求模型用于校验客户端提交的数据，响应模型用于规定接口返回的字段。

特别需要注意：ArticleUpdate 中使用了可选字段。后续更新接口应该使用 model_dump(exclude_unset=True)，以区分客户端未提交字段和明确提交的字段。

八、实现 JWT 身份认证

创建 app/security.py：

import os
from datetime import datetime, timedelta, timezone

import jwt
from pwdlib import PasswordHash

JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = "HS256"
TOKEN_EXPIRE_MINUTES = 30

password_hash = PasswordHash.recommended()


def hash_password(password: str) -> str:
    return password_hash.hash(password)


def verify_password(
    password: str,
    hashed_password: str,
) -> bool:
    return password_hash.verify(
        password,
        hashed_password,
    )


def create_access_token(user_id: int) -> str:
    now = datetime.now(timezone.utc)

    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + timedelta(
            minutes=TOKEN_EXPIRE_MINUTES
        ),
    }

    return jwt.encode(
        payload,
        JWT_SECRET_KEY,
        algorithm=JWT_ALGORITHM,
    )


def decode_access_token(token: str) -> dict:
    return jwt.decode(
        token,
        JWT_SECRET_KEY,
        algorithms=[JWT_ALGORITHM],
        options={"require": ["sub", "exp"]},
    )

通过环境变量设置 JWT 密钥：

export JWT_SECRET_KEY="$(python -c 'import secrets; print(secrets.token_hex(32))')"

这条命令适合在首次配置时生成密钥。生产环境必须固定保存生成的密钥，不能在每次启动应用时重新生成。

JWT 载荷不应存放密码、密码哈希等敏感信息。签名令牌的载荷通常可以被解码读取。

九、实现登录与当前用户依赖

创建 app/deps.py：

import jwt

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.database import get_db
from app.models import User
from app.security import decode_access_token

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="/auth/login"
)


def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db),
) -> User:
    error = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="登录凭证无效或已过期",
        headers={"WWW-Authenticate": "Bearer"},
    )

    try:
        payload = decode_access_token(token)
        subject = payload.get("sub")

        if not isinstance(subject, str) or not subject.isdigit():
            raise error

        user_id = int(subject)

    except (jwt.InvalidTokenError, ValueError):
        raise error

    user = db.get(User, user_id)

    if user is None or not user.is_active:
        raise error

    return user


def get_current_admin(
    current_user: User = Depends(get_current_user),
) -> User:
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="权限不足",
        )

    return current_user

这里将身份认证与权限控制分开：

• get_current_user()：确认用户身份。

• get_current_admin()：确认当前用户是否拥有管理员权限。

后续所有需要登录的接口都可以通过依赖注入复用这套逻辑。

创建 app/routers/auth.py：

import jwt

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.database import get_db
from app.models import User
from app.schemas import TokenResponse
from app.security import create_access_token, verify_password

router = APIRouter(prefix="/auth", tags=["认证"])


@router.post("/login", response_model=TokenResponse)
def login(
    form: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db),
):
    user = db.scalar(
        select(User).where(
            User.username == form.username.strip()
        )
    )

    if user is None or not verify_password(
        form.password,
        user.hashed_password if user else "",
    ):
        raise HTTPException(
            status_code=401,
            detail="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )

    if not user.is_active:
        raise HTTPException(
            status_code=403,
            detail="账号已停用",
        )

    return TokenResponse(
        access_token=create_access_token(user.id)
    )

该登录接口接收表单格式的用户名和密码，而不是 JSON 格式。如果客户端需要提交 JSON 登录请求，可以自行设计对应的 Pydantic 请求模型，但要确保 Swagger OAuth2 登录流程与接口格式保持一致。

十、实现文章管理接口

创建 app/routers/articles.py：

from fastapi import (
    APIRouter,
    Depends,
    HTTPException,
    Query,
    status,
)
from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.database import get_db
from app.deps import get_current_admin
from app.models import Article, Category, User
from app.schemas import (
    ArticleCreate,
    ArticlePage,
    ArticleResponse,
    ArticleUpdate,
)

router = APIRouter(prefix="/articles", tags=["文章管理"])


def validate_status(value: str) -> str:
    if value not in {"draft", "published"}:
        raise HTTPException(
            status_code=422,
            detail="文章状态必须为 draft 或 published",
        )
    return value


@router.post(
    "",
    response_model=ArticleResponse,
    status_code=status.HTTP_201_CREATED,
)
def create_article(
    data: ArticleCreate,
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    article_status = validate_status(data.status)

    if data.category_id is not None:
        category = db.get(Category, data.category_id)
        if category is None:
            raise HTTPException(
                status_code=404,
                detail="分类不存在",
            )

    article = Article(
        title=data.title,
        slug=data.slug,
        content=data.content,
        status=article_status,
        category_id=data.category_id,
        author_id=admin.id,
    )

    db.add(article)

    try:
        db.commit()
        db.refresh(article)
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=409,
            detail="文章标识已存在或数据约束不满足",
        )

    return article


@router.get("", response_model=ArticlePage)
def list_articles(
    keyword: str | None = Query(None, max_length=100),
    category_id: int | None = Query(None, ge=1),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    conditions = []

    if keyword and keyword.strip():
        pattern = f"%{keyword.strip()}%"
        conditions.append(
            or_(
                Article.title.like(pattern),
                Article.content.like(pattern),
            )
        )

    if category_id is not None:
        conditions.append(
            Article.category_id == category_id
        )

    total = db.scalar(
        select(func.count())
        .select_from(Article)
        .where(*conditions)
    ) or 0

    statement = (
        select(Article)
        .where(*conditions)
        .order_by(
            Article.created_at.desc(),
            Article.id.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
    )

    items = db.scalars(statement).all()

    return {
        "total": total,
        "page": page,
        "page_size": page_size,
        "items": items,
    }


@router.get("/{article_id}", response_model=ArticleResponse)
def get_article(
    article_id: int,
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    article = db.get(Article, article_id)

    if article is None:
        raise HTTPException(
            status_code=404,
            detail="文章不存在",
        )

    return article


@router.patch("/{article_id}", response_model=ArticleResponse)
def update_article(
    article_id: int,
    data: ArticleUpdate,
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    article = db.get(Article, article_id)

    if article is None:
        raise HTTPException(
            status_code=404,
            detail="文章不存在",
        )

    changes = data.model_dump(exclude_unset=True)

    if "status" in changes and changes["status"] is not None:
        changes["status"] = validate_status(changes["status"])

    if (
        "category_id" in changes
        and changes["category_id"] is not None
        and db.get(Category, changes["category_id"]) is None
    ):
        raise HTTPException(
            status_code=404,
            detail="分类不存在",
        )

    for field, value in changes.items():
        setattr(article, field, value)

    try:
        db.commit()
        db.refresh(article)
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=409,
            detail="文章标识已存在或数据约束不满足",
        )

    return article


@router.delete("/{article_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_article(
    article_id: int,
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    article = db.get(Article, article_id)

    if article is None:
        raise HTTPException(
            status_code=404,
            detail="文章不存在",
        )

    db.delete(article)
    db.commit()
    return None

文章接口说明

以上接口均要求管理员登录，便于演示权限控制。

实际内容管理系统通常需要区分公开内容 API 与后台管理 API。公开列表只返回已发布文章，后台接口则允许管理员查看草稿、编辑内容和删除文章。

为什么使用 PATCH 更新文章？

PATCH 适合只修改部分字段。例如只更新标题时，不需要重新提交正文和分类。

这里使用：

data.model_dump(exclude_unset=True)

它只包含客户端明确提交的字段，可以避免未提交字段意外覆盖数据库中的原有数据。

如果允许客户端显式将某个可空字段设置为 null，则需要根据业务规则进一步验证字段是否允许清空。

十一、实现分类管理接口

创建 app/routers/categories.py：

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.database import get_db
from app.deps import get_current_admin
from app.models import Category, User
from app.schemas import CategoryCreate, CategoryResponse

router = APIRouter(
    prefix="/categories",
    tags=["分类管理"],
)


@router.post(
    "",
    response_model=CategoryResponse,
    status_code=201,
)
def create_category(
    data: CategoryCreate,
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    category = Category(
        name=data.name.strip(),
        slug=data.slug.strip(),
    )

    if not category.name or not category.slug:
        raise HTTPException(
            status_code=422,
            detail="分类名称和标识不能为空",
        )

    db.add(category)

    try:
        db.commit()
        db.refresh(category)
    except IntegrityError:
        db.rollback()
        raise HTTPException(
            status_code=409,
            detail="分类名称或标识已经存在",
        )

    return category


@router.get("", response_model=list[CategoryResponse])
def list_categories(
    db: Session = Depends(get_db),
    admin: User = Depends(get_current_admin),
):
    return db.scalars(
        select(Category).order_by(Category.id.desc())
    ).all()

这样就完成了分类的创建和查询。

由于分类数量通常远少于文章数量，初期可以直接返回全部分类。如果分类规模较大，再增加分页、搜索和缓存。

十二、注册路由并创建应用

最后修改 app/main.py：

from fastapi import FastAPI

from app.database import Base, engine
from app.routers import articles, auth, categories

app = FastAPI(
    title="FastAPI CMS",
    description="基于 FastAPI 和 MySQL 的内容管理 API",
    version="1.0.0",
)

# 仅用于本地教学演示；正式项目使用 Alembic 迁移。
Base.metadata.create_all(bind=engine)

app.include_router(auth.router)
app.include_router(articles.router)
app.include_router(categories.router)


@app.get("/health", tags=["系统"])
def health_check():
    return {"status": "ok"}

运行：

uvicorn app.main:app --reload

打开：

http://127.0.0.1:8000/docs

FastAPI 会自动生成 API 文档，可以直接测试各个接口。

需要注意：Base.metadata.create_all() 只能帮助创建不存在的表，不负责管理后续复杂的表结构变更。生产项目应使用 Alembic 管理数据库迁移。

十三、创建首个管理员账号

为了避免任何人都可以注册管理员账号，本教程没有提供公开的管理员注册接口。

可以创建一次性的初始化脚本，在受控的服务器环境中创建第一个管理员。

例如，在项目根目录创建 create_admin.py：

import getpass

from sqlalchemy import select

from app.database import SessionLocal
from app.models import User
from app.security import hash_password


def main():
    username = input("管理员用户名：").strip()
    password = getpass.getpass("管理员密码：")

    if len(username) < 3:
        raise SystemExit("用户名至少需要 3 个字符")

    if len(password) < 12:
        raise SystemExit("初始化管理员密码至少需要 12 个字符")

    with SessionLocal() as db:
        existing = db.scalar(
            select(User).where(
                User.username == username
            )
        )

        if existing is not None:
            raise SystemExit("该用户名已经存在")

        admin = User(
            username=username,
            hashed_password=hash_password(password),
            is_active=True,
            is_admin=True,
        )

        db.add(admin)
        db.commit()

    print("管理员账号创建成功")


if __name__ == "__main__":
    main()

在已正确设置环境变量的项目目录运行：

python -m create_admin

请不要把这个脚本作为公开 HTTP 接口使用，也不要将真实管理员密码写入源代码或版本控制系统。

十四、测试完整工作流程

建议按照下面的顺序进行测试。

1. 创建管理员账号

运行初始化脚本，创建管理员。

2. 登录获取令牌

在 Swagger 中调用：

POST /auth/login

提交用户名和密码，获取 access_token。

3. 在 Swagger 中授权

点击 Swagger 页面中的 Authorize，按界面要求输入访问令牌。如果界面使用 OAuth2 登录流程，也可以直接通过登录接口完成授权。

之后调用需要管理员权限的接口时，Swagger 会自动携带令牌。

4. 创建分类

调用：

POST /categories

请求体示例：

{
  "name": "技术教程",
  "slug": "technology"
}

5. 创建文章

调用：

POST /articles

请求体示例：

{
  "title": "FastAPI CMS 开发教程",
  "slug": "fastapi-cms-guide",
  "content": "本文介绍如何使用 FastAPI 构建内容管理系统。",
  "category_id": 1,
  "status": "draft"
}

6. 搜索文章

调用：

GET /articles?keyword=FastAPI&page=1&page_size=20

接口将返回匹配的文章、总记录数和分页参数。

7. 编辑和删除

通过文章 ID 调用 PATCH 或 DELETE 接口。

测试时还应验证未登录请求会返回 401，普通用户请求管理员接口会返回 403，不存在的文章返回 404，重复的 slug 返回 409。

十五、生产环境还需要完善哪些功能？

本文完成的是一个具备基本内容管理能力的 CMS 后端，不等于已经具备所有生产环境功能。

正式上线前，建议继续完善以下部分。

1. 数据库迁移

引入 Alembic，管理表结构变更，避免直接修改线上数据库结构。

2. 权限与数据范围

如果将来支持多个作者，需要区分“管理员可以编辑所有文章”和“作者只能编辑自己的文章”。仅验证用户已经登录是不够的。

3. 公开内容接口

单独设计公开文章列表和详情接口，默认只展示已发布内容，不能让匿名访客通过后台接口读取草稿。

4. HTML 内容安全

如果文章内容允许保存 HTML，需要对输入进行适当的清理或使用安全的渲染策略，避免存储型 XSS。

5. 日志与审计

记录管理员登录、内容发布、删除和权限变更等重要操作。日志不应包含密码、JWT 原文或其他敏感凭据。

6. 搜索性能

当文章数量增长到几十万或百万级时，应评估 MySQL 索引、全文搜索以及专用搜索引擎，而不是无限制地使用 LIKE '%关键词%'。

7. 媒体管理

图片和音频应通过专门的上传、校验和存储流程管理。大文件可以考虑对象存储或独立静态资源服务，避免将文件直接塞进数据库。

8. 部署与备份

使用 Debian、Nginx、systemd 部署应用，并建立数据库备份、恢复演练、HTTPS 证书续期和监控机制。

9. FastAPI 接口访问限制

正式上线时，应根据接口用途限制访问范围，避免后台管理接口、调试接口和内部接口被未授权人员调用。

• 关闭或限制 API 文档： 生产环境可以关闭 Swagger UI、ReDoc 和 OpenAPI Schema，或者仅允许经过授权的管理员访问，减少接口结构暴露。

• 保护后台管理接口： 对文章新增、修改、删除、分类管理和用户管理等接口，实施身份认证与权限校验，不能仅依赖前端隐藏按钮。

• 限制内部接口访问： 如果某些接口只供内部服务调用，可以通过 Nginx、网络防火墙、IP 白名单或独立网络限制访问来源。

• 增加请求频率限制： 对登录、注册、搜索等容易被滥用的接口设置合理的限流策略，降低暴力尝试和恶意请求的影响。

• 统一处理未授权请求： 未登录或身份验证失败时返回 401 Unauthorized；已登录但权限不足时返回 403 Forbidden。

• 不要将 URL 隐蔽当作安全措施： 修改接口路径、隐藏前端入口或关闭 Swagger，都不能代替服务端的身份认证和权限校验。

例如，可以在创建 FastAPI 应用时关闭默认的 API 文档入口：

from fastapi import FastAPI

app = FastAPI(
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
)

这会同时关闭 Swagger UI、ReDoc 和 OpenAPI Schema 的默认路由。注意，这并不会屏蔽实际业务 API，也不会自动保护已有接口。真正的访问控制仍然需要在接口依赖、认证授权和服务器网络配置中实现。

如果需要保留开发环境的 API 文档，可以根据环境变量区分开发和生产配置，而不是在所有环境中一律关闭。

十六、总结

通过本教程，我们将 FastAPI 的多个基础知识整合为一个 CMS 后端，完成了：

• 项目模块化组织。

• MySQL 数据库模型设计。

• JWT 登录与管理员认证。

• 文章创建、查询、更新和删除。

• 分类创建与查询。

• 分页和关键词搜索。

• 请求验证和统一响应模型。

• 首个管理员初始化。

• 生产环境扩展规划。

这套结构适合作为个人技术博客、内容管理后台网站的起点。