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

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


1. RESTful 设计原则

1.1 URL 设计

传统:
GET  /get_articles      → ❌ URL 包含动词
POST /do_login          → ❌ 动词

RESTful:
GET    /api/articles     → 获取文章列表
POST   /api/articles     → 创建文章
GET    /api/articles/42 → 获取 ID 为 42 的文章
PUT    /api/articles/42 → 更新文章
DELETE /api/articles/42 → 删除文章

1.2 HTTP 状态码

状态码含义适用场景
200OK成功获取/更新资源
201Created成功创建资源
204No Content成功删除(无返回体)
400Bad Request请求参数错误
401Unauthorized未登录
403Forbidden无权限
404Not Found资源不存在
422Unprocessable Entity验证失败
500Internal Server Error服务器错误

2. 基础 JSON API

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

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

from . import articles, auth, users
# 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():
    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)

    query = Article.query.filter_by(is_published=True)
    if category_id:
        query = query.filter_by(category_id=category_id)

    pagination = query.order_by(Article.created_at.desc())\
        .paginate(page=page, per_page=per_page, error_out=False)

    return jsonify({
        "articles": [
            {
                "id": a.id,
                "title": a.title,
                "summary": a.summary,
                "author": {"id": a.author.id, "name": a.author.username},
                "views": a.views,
                "created_at": a.created_at.isoformat(),
            }
            for a in pagination.items
        ],
        "total": pagination.total,
        "page": page,
        "pages": pagination.pages,
        "has_next": pagination.has_next,
        "has_prev": pagination.has_prev,
    })

@api_bp.route("/articles/<int:article_id>", methods=["GET"])
def get_article(article_id):
    article = Article.query.get_or_404(article_id)
    article.views += 1
    db.session.commit()

    return jsonify({
        "id": article.id,
        "title": article.title,
        "content": article.content,
        "content_html": render_markdown(article.content),
        "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
def create_article():
    data = request.get_json()
    if not data:
        abort(400, description="请求体不能为空")

    form = ArticleForm(data=data)
    if not form.validate():
        return jsonify({"errors": form.errors}), 422

    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()

    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()
    for field in ["title", "content", "summary", "is_published"]:
        if field in data:
            setattr(article, field, data[field])

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

@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()
    return "", 204

3. API 错误处理

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

def api_error(code, message, details=None):
    response = jsonify({"error": message})
    if details:
        response["details"] = details
    return response, code

# 全局注册
@api_bp.errorhandler(404)
def not_found(e):
    return api_error(404, "资源不存在")

@api_bp.errorhandler(403)
def forbidden(e):
    return api_error(403, "权限不足")

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

@api_bp.errorhandler(500)
def server_error(e):
    return api_error(500, "服务器内部错误")

4. 小结

# RESTful API 速查

# 返回 JSON
return jsonify({"key": "value"})

# 状态码
return jsonify({}), 201   # 创建成功
return jsonify({}), 204  # 删除成功
return jsonify({}), 422  # 验证失败

# 错误
abort(404)  # 资源不存在
abort(403)  # 无权访问

💡 设计原则:URL 表示资源(名词复数),HTTP 方法表示操作。状态码要准确返回,错误信息要清晰。


🔗 扩展阅读