RESTful API 开发:为 App/小程序提供轻量 JSON 接口

📂 所属阶段:第五阶段 — 高级进阶(性能与架构)
🔗 相关章节:Flask 上下文深挖 · 数据验证

0. 为什么选 REST + Flask?

现在的前后端分离架构已经成为主流:无论是移动 App、微信小程序,还是 Vue/React 单页应用,都需要一套统一、无状态、以资源为中心的后端接口。RESTful 风格正好满足这些要求,而 Flask 轻量灵活的特点,又让我们能快速搭出规范化的接口原型,再平滑迭代到生产可用。

本文会通过一个完整的「文章管理 API」案例,带你一步步掌握用 Flask 开发 RESTful JSON 接口的核心技巧。

1. RESTful 设计核心:资源 + 状态码

1.1 URL 只能是「资源名词」

URL 的唯一任务是标识资源,里面不应该出现“获取、创建、删除”这类动作词——这些操作统统交给 HTTP 方法去表达。

// ❌ 反例:动词出现在 URL 里
GET /get_articles          → 动词 get
POST /do_login             → 动词 do
GET /getArticleDetail?id=42 → 资源层级不清晰

// ✅ 正例:RESTful 的规范 URL
GET    /api/articles        → 获取已发布文章列表(支持分页和筛选)
POST   /api/articles        → 创建一篇新文章
GET    /api/articles/42     → 获取 id 为 42 的文章详情
PUT    /api/articles/42     → 全量更新 id 为 42 的文章
DELETE /api/articles/42     → 删除 id 为 42 的文章

💡 几个实用细节:

  • 统一用 /api 前缀,方便区分后端接口和前端路由。
  • 资源名用复数形式(articles 而不是 article),因为列表是资源的集合。
  • 层级不要超过 3 层(例如 /api/articles/42/comments 没问题,但 /api/users/1/categories/5/articles 就过于复杂,那时可以用查询参数或反向关联接口代替)。

1.2 准确使用 HTTP 状态码

状态码是人和机器都能读懂的响应标记,千万别所有情况都返回 200500。这里列出 RESTful 开发中最常用的几个:

状态码英文名称使用场景
200OK查询或全量更新成功,并返回了数据
201Created创建新资源成功,通常返回新资源的标识(如 id)
204No Content删除、清空等操作成功,响应体必须为空
400Bad Request请求参数格式错误(比如 JSON 解析失败、缺少必填字段)
401Unauthorized未登录或令牌过期(身份凭证无效)
403Forbidden已登录但没有操作权限
404Not Found要访问的资源不存在,或路由写错了
422Unprocessable Entity参数格式正确但业务校验不通过(比如用户名太短、分类不存在)
500Internal Server Error服务器代码出错(生产环境中不要把错误详情返回给前端!)

2. 用 Flask 实现基础文章管理 API

这里假设项目中已经具备了用户认证(Flask-Login)、数据库(SQLAlchemy)和表单验证(Flask-WTF 或 Marshmallow)的基础,本文只集中讲解 API 部分的核心代码。

2.1 先创建 API 蓝图

用蓝图把接口和页面路由、模板拆分开,更符合模块化开发的习惯:

# app/api/__init__.py
from flask import Blueprint

# 加上 url_prefix=/api,后面所有子路由都会自动带上这个前缀
api_bp = Blueprint("api", __name__, url_prefix="/api")

# 延迟导入子路由,避免循环引用
from . import articles, auth, users

2.2 完整的文章 CRUD 接口

下面是文章管理 API 的全部实现,包含分页列表、详情、创建、更新和删除。

# app/api/articles.py
from flask import jsonify, request, abort
from flask_login import login_required, current_user
from app.api import api_bp
from app.extensions import db
from app.models import Article, Category
from app.forms import ArticleForm
from app.utils import render_markdown


# ------------------------------
# 获取文章列表(分页 + 分类筛选)
# ------------------------------
@api_bp.route("/articles", methods=["GET"])
def list_articles():
    # 1. 从查询参数中提取输入
    page = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", 20, type=int)
    category_id = request.args.get("category", type=int)  # 可选筛选条件

    # 2. 只查询已发布的文章
    query = Article.query.filter_by(is_published=True)
    if category_id:
        query = query.filter_by(category_id=category_id)

    # 3. 按时间倒序,然后分页(error_out=False 表示超过总页数时返回空列表,而不是 404)
    pagination = query.order_by(Article.created_at.desc()) \
        .paginate(page=page, per_page=per_page, error_out=False)

    # 4. 组装 JSON 响应
    return jsonify({
        "data": [
            {
                "id": a.id,
                "title": a.title,
                "summary": a.summary,
                "author": {"id": a.author.id, "名称": a.author.username},
                "views": a.views,
                "created_at": a.created_at.isoformat(),  # 统一使用 ISO8601 时间格式
            }
            for a in pagination.items
        ],
        "pagination": {
            "total": pagination.total,      # 总文章数
            "page": page,                  # 当前页
            "pages": pagination.pages,     # 总页数
            "has_next": pagination.has_next, # 是否有下一页
            "has_prev": pagination.has_prev, # 是否有上一页
        }
    })


# ------------------------------
# 获取单篇文章详情(阅读量自动 +1)
# ------------------------------
@api_bp.route("/articles/<int:article_id>", methods=["GET"])
def get_article(article_id):
    # 1. 查不到直接返回 404
    article = Article.query.get_or_404(article_id)

    # 2. 业务逻辑:每次查看详情,阅读数加 1
    article.views += 1
    db.session.commit()

    # 3. 返回详细信息(包含 Markdown 渲染后的 HTML)
    return jsonify({
        "id": article.id,
        "title": article.title,
        "content": article.content,
        "content_html": render_markdown(article.content),  # 把 Markdown 转成 HTML
        "summary": article.summary,
        "author": {
            "id": article.author.id,
            "name": article.author.username,
            "avatar": article.author.get_avatar(),
        },
        "category": {"id": article.category.id, "name": article.category.name} if article.category else None,
        "views": article.views,
        "created_at": article.created_at.isoformat(),
        "updated_at": article.updated_at.isoformat(),
    })


# ------------------------------
# 创建新文章(需要登录)
# ------------------------------
@api_bp.route("/articles", methods=["POST"])
@login_required  # Flask-Login 装饰器,确保只有已登录用户才能访问
def create_article():
    # 1. 检查请求体是否为 JSON
    data = request.get_json()
    if not data:
        abort(400, description="请求体必须是有效的 JSON 格式")

    # 2. 对数据进行验证(这里用 Flask-WTF,也可以换成 Marshmallow)
    form = ArticleForm(data=data)
    if not form.validate():
        return jsonify({"error": "数据验证失败", "details": form.errors}), 422

    # 3. 创建数据并存入数据库
    article = Article(
        title=form.title.data,
        content=form.content.data,
        summary=form.summary.data,
        category_id=form.category_id.data or None,
        author=current_user,
    )
    db.session.add(article)
    db.session.commit()

    # 4. 返回 201 和新创建的 id
    return jsonify({"id": article.id, "message": "文章创建成功"}), 201


# ------------------------------
# 全量/部分更新文章(需要登录且有权限)
# ------------------------------
@api_bp.route("/articles/<int:article_id>", methods=["PUT"])
@login_required
def update_article(article_id):
    article = Article.query.get_or_404(article_id)

    # 权限检查:只有作者本人或管理员能修改
    if article.author_id != current_user.id and not current_user.is_admin:
        abort(403, description="无权修改此文章")

    data = request.get_json()
    if not data:
        abort(400, description="请求体不能为空")

    # 允许更新的字段白名单(用部分更新兼容 PUT / PATCH)
    allowed_fields = ["title", "content", "summary", "is_published", "category_id"]
    for field in allowed_fields:
        if field in data:
            setattr(article, field, data[field])

    db.session.commit()
    return jsonify({"message": "文章更新成功"}), 200


# ------------------------------
# 删除文章(需要登录且有权限)
# ------------------------------
@api_bp.route("/articles/<int:article_id>", methods=["DELETE"])
@login_required
def delete_article(article_id):
    article = Article.query.get_or_404(article_id)

    if article.author_id != current_user.id and not current_user.is_admin:
        abort(403)

    db.session.delete(article)
    db.session.commit()

    # 删除成功必须返回 204,且不能带响应体
    return "", 204

3. 统一的 API 错误处理

上面的代码中,我们直接使用了 abort() 和手动返回状态码。但 Flask 默认的错误响应是 HTML 格式的,这对 API 很不友好。我们需要给 API 蓝图注册专门的 JSON 错误处理器

# app/api/errors.py
from flask import jsonify
from app.api import api_bp


def _api_error_response(code: int, message: str, details=None):
    """
    生成标准化的 JSON 错误响应
    :param code: HTTP 状态码
    :param message: 给前端展示的简洁错误提示
    :param details: 详细的错误信息(可选,生产环境可以隐藏)
    """
    body = {"error": message}
    # 开发阶段可返回细节,生产环境建议去掉下面这行
    if details:
        body["details"] = details
    return jsonify(body), code


@api_bp.errorhandler(400)
def bad_request(e):
    return _api_error_response(400, str(e.description) or "请求参数格式错误")


@api_bp.errorhandler(401)
def unauthorized(e):
    return _api_error_response(401, "未登录或登录已过期,请重新登录")


@api_bp.errorhandler(403)
def forbidden(e):
    return _api_error_response(403, str(e.description) or "权限不足,无法执行此操作")


@api_bp.errorhandler(404)
def not_found(e):
    return _api_error_response(404, str(e.description) or "请求的资源不存在")


@api_bp.errorhandler(422)
def validation_error(e):
    return _api_error_response(422, "数据验证失败", details=e.description)


@api_bp.errorhandler(500)
def server_error(e):
    # 生产环境绝对不要返回具体的错误信息!
    return _api_error_response(500, "服务器内部错误,请稍后重试")

💡 别忘了在 app/api/__init__.py 里导入错误处理模块,否则这些处理器不会生效:

# app/api/__init__.py(修改后)
from flask import Blueprint

api_bp = Blueprint("api", __name__, url_prefix="/api")

from . import articles, auth, users
from . import errors  # 放在子路由导入后面,触发注册

4. 核心要点速查

4.1 常用 Flask API 快捷方法

代码片段作用
jsonify(dict/list)把 Python 对象转为 JSON 格式的响应
request.get_json()获取请求体里的 JSON 数据(不是 JSON 时返回 None
request.args.get(key, default, type)获取 URL 查询参数,比如 ?page=2 中的 page
abort(code, description)中断请求,直接返回指定的错误状态码
Model.query.paginate()SQLAlchemy 的分页查询方法

4.2 RESTful 设计避坑指南

  1. 可以加版本号前缀:比如 /api/v1/articles,方便接口升级时不影响旧版客户端。
  2. 别只用 GET 和 POST:根据操作选择合适的 HTTP 方法(GET/POST/PUT/DELETE/PATCH)。
  3. 永远不要向 API 返回 HTML:所有响应都应该是 JSON。
  4. 统一时间格式:建议使用 ISO8601(例如 2024-05-20T12:34:56Z 或带时区的本地时间)。
  5. 分页、排序、筛选都放在查询参数里:比如 ?page=1&sort=created_at
  6. 删除成功后必须返回 204,且不能带响应体

5. 扩展方向

如果要把这个 API 升级成真正生产可用的服务,可以继续学习以下内容:

  1. 接口文档自动生成:使用 Flask-RESTXFlask-APISpec 直接生成 Swagger 文档。
  2. JWT 令牌认证:Flask-Login 依赖 Session,适合 Web,而移动 App 和小程序更适合用 Flask-JWT-Extended 实现无状态认证。
  3. 接口限流:通过 Flask-Limiter 防止恶意刷请求。
  4. 数据验证优化:用 Marshmallow 取代 Flask-WTF,它更专注于 API 的序列化/反序列化,不需要 CSRF 保护。
  5. API 缓存:用 Flask-Caching 缓存高频查询(如文章列表),降低数据库压力。

🔗 扩展阅读