RESTful API 设计与实现

第 7 章 · RESTful API 设计与实现

本章目标：在 api-demo 中设计并实现 RESTful JSON API；使用 jsonify 与 Blueprint 组织路由；实现分页、统一错误码与 CORS 跨域；完成与前端 SPA 的 JSON 对接；在 ch05 Product 模型与 ch06 登录基础上暴露商品与用户接口。

学时建议：4～5 小时（含 2 小时 API 联调）

前置：本模块 ch01 应用工厂；ch02 蓝图与请求上下文；ch05 Flask-SQLAlchemy 与 Product 模型；ch06 Flask-Login 会话。JSON 与 requests 见 python-dev ch11。

后续章节：ch08 将在现有模型上增量添加 cover、description 等字段，请勿整段替换 Product 定义。

7.1 为什么需要 REST API

前后端分离架构中，Flask 后端不再返回 HTML，而是提供 JSON 供 Vue/React 或移动端消费。

浏览器 / SPA 前端
    → HTTP GET/POST/PUT/PATCH/DELETE
    → Flask Blueprint (api)
    → SQLAlchemy Model ↔ 数据库
    → jsonify → JSON Response

说明：本章示例均基于教学项目 api-demo，API 基址使用 https://api.example.com 或本地 http://127.0.0.1:5000，不引用任何企业内部仓库或私有网关。

7.2 项目结构与 API 蓝图

在 ch01 应用工厂基础上，新增 api 蓝图：

api-demo/
├── api_demo/
│   ├── __init__.py          # create_app()
│   ├── extensions.py        # db, login_manager
│   └── config.py
├── api_demo/api/
│   ├── __init__.py          # api_bp = Blueprint("api", __name__)
│   ├── products.py
│   ├── errors.py
│   └── schemas.py           # 序列化辅助（可选 marshmallow）
├── api_demo/models/
│   └── product.py
└── run.py

api_demo/api/__init__.py：

from flask import Blueprint

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

from . import products, errors  # noqa: E402, F401

api_demo/__init__.py 注册：

def create_app(config_name=None):
    app = Flask(__name__)
    # ... 加载配置、初始化 db、login_manager ...
    from api_demo.api import api_bp
    app.register_blueprint(api_bp)
    return app

7.3 jsonify 与响应结构

Flask 内置 jsonify 自动设置 Content-Type: application/json。中文默认不转义需关闭 ASCII 编码——Flask 2.3+/3.x 已移除 JSON_AS_ASCII 配置键，应改用 JSON provider：

在应用工厂（api_demo/__init__.py）中：

def create_app(config_name="development"):
    app = Flask(__name__)
    # ...
    app.json.ensure_ascii = False   # 中文不转义为 \uXXXX
    app.json.sort_keys = False      # 保持字段声明顺序
    return app

旧教程里的 JSON_AS_ASCII = False / JSON_SORT_KEYS = False 配置键在 Flask 3.x 中已失效，请勿再写进 config.py。

统一成功响应（推荐 envelope 模式）：

# api_demo/api/utils.py
from flask import jsonify


def ok(data=None, message="ok", status=200, **extra):
  payload = {"code": 0, "message": message, "data": data}
  payload.update(extra)
  return jsonify(payload), status

商品序列化（避免直接 jsonify(model)，手动转 dict 可控字段）：

# api_demo/api/schemas.py
def product_to_dict(product):
    return {
        "id": product.id,
        "name": product.name,
        "slug": product.slug,
        "price": str(product.price),      # Decimal → 字符串避免精度问题
        "stock": product.stock,
        "is_published": product.is_published,
        "created_at": product.created_at.isoformat() if product.created_at else None,
    }

products.py 列表接口：

from flask import request
from api_demo.extensions import db
from api_demo.models.product import Product
from . import api_bp
from .schemas import product_to_dict
from .utils import ok


@api_bp.get("/products")
def list_products():
    page = request.args.get("page", 1, type=int)
    per_page = min(request.args.get("per_page", 10, type=int), 50)
    q = Product.query.filter_by(is_published=True).order_by(Product.id.desc())
    pagination = q.paginate(page=page, per_page=per_page, error_out=False)
    items = [product_to_dict(p) for p in pagination.items]
    return ok(
        data=items,
        pagination={
            "page": pagination.page,
            "per_page": pagination.per_page,
            "total": pagination.total,
            "pages": pagination.pages,
            "has_next": pagination.has_next,
            "has_prev": pagination.has_prev,
        },
    )

7.4 分页参数与 SQLAlchemy paginate

前端请求示例：

GET /api/v1/products?page=2&per_page=20 HTTP/1.1
Host: api.example.com
Accept: application/json

响应示例：

{
  "code": 0,
  "message": "ok",
  "data": [
    {"id": 21, "name": "Flask 入门", "slug": "flask-intro", "price": "49.00", "stock": 100, "is_published": true}
  ],
  "pagination": {
    "page": 2,
    "per_page": 20,
    "total": 45,
    "pages": 3,
    "has_next": true,
    "has_prev": true
  }
}

7.5 CRUD 与 HTTP 方法

@api_bp.get("/products/<int:product_id>")
def get_product(product_id):
    product = Product.query.get_or_404(product_id)
    return ok(data=product_to_dict(product))


@api_bp.post("/products")
@login_required
def create_product():
    body = request.get_json(silent=True) or {}
    name = (body.get("name") or "").strip()
    if not name:
        return fail(42201, "name 不能为空", status=422)
    product = Product(
        name=name,
        slug=body.get("slug") or slugify(name),
        price=body.get("price", 0),
        stock=body.get("stock", 0),
    )
    db.session.add(product)
    db.session.commit()
    return ok(data=product_to_dict(product), message="created", status=201)