OpenAPI 与 Flask-RESTX / Swagger

第 14 章 · OpenAPI 与 Flask-RESTX / Swagger

本章目标：理解 OpenAPI 3 规范与 REST API 文档的价值；使用 Flask-RESTX 的 Api/Namespace/Resource 自动生成 Swagger UI；了解 apispec + flask-smorest 简要对比；为 api-demo 的 /api/v1 提供可交互文档；将请求/响应模型与错误码文档化。

学时建议：4～5 小时（含 2 小时文档站联调）

前置：本模块 ch01～ch13；重点复习 ch07 REST 路由与 envelope 响应、ch13 ProductSchema 字段定义。

14.1 为什么需要 API 文档

┌─────────────┐     OpenAPI 契约      ┌─────────────┐
│  user-demo   │ ◄──────────────────► │  api-demo    │
│  前端 SPA    │   路径/参数/响应模型   │  Flask API   │
└─────────────┘                       └─────────────┘
         └──────── Swagger UI 在线调试 ──────────┘

OpenAPI 3 核心结构：paths（路径与方法）、components/schemas（模型）、responses（状态码）、securitySchemes（Bearer JWT，ch15）。

示例域名：api.example.com；前端：user-demo；禁止写入真实生产密钥。

14.2 Flask-RESTX 初始化

pip install flask-restx

# api_demo/api/restx_init.py
from flask_restx import Api

authorizations = {
    "Bearer": {
        "type": "apiKey", "in": "header", "name": "Authorization",
        "description": "格式: Bearer <access_token>",
    }
}

api = Api(
    title="api-demo API",
    version="1.0",
    description="教学项目 · 商品与用户 REST API",
    doc="/docs",
    prefix="/api/v1",
    authorizations=authorizations,
    security="Bearer",
)

# api_demo/__init__.py
def create_app(config_name=None):
    app = Flask(__name__)
    from api_demo.api.restx_init import api
    from api_demo.api import namespaces  # noqa: F401
    api.init_app(app)
    return app

启动后访问 http://127.0.0.1:5000/docs。

14.3 Namespace 与模型定义

# api_demo/api/namespaces/products_ns.py
from flask_restx import Namespace, fields

ns = Namespace("products", description="商品资源")

product_model = ns.model("Product", {
    "id": fields.Integer(readOnly=True, example=1),
    "name": fields.String(required=True, example="Flask 入门实战"),
    "slug": fields.String(example="flask-intro"),
    "price": fields.String(example="49.00"),
    "stock": fields.Integer(example=100),
    "is_published": fields.Boolean(example=True),
    "created_at": fields.DateTime(readOnly=True),
})

product_input = ns.model("ProductInput", {
    "name": fields.String(required=True),
    "slug": fields.String,
    "price": fields.String(required=True, description="十进制字符串"),
    "stock": fields.Integer(default=0),
    "is_published": fields.Boolean(default=False),
})

envelope_list = ns.model("ProductListEnvelope", {
    "code": fields.Integer(example=0),
    "message": fields.String(example="ok"),
    "data": fields.List(fields.Nested(product_model)),
    "pagination": fields.Nested(ns.model("Pagination", {
        "page": fields.Integer, "per_page": fields.Integer,
        "total": fields.Integer, "pages": fields.Integer,
    })),
})

error_model = ns.model("ErrorEnvelope", {
    "code": fields.Integer(example=42201),
    "message": fields.String(example="参数校验失败"),
    "errors": fields.Raw,
})

注册：api.add_namespace(products_ns)（在 namespaces/__init__.py）。

14.4 Resource 与装饰器

from flask import request
from flask_restx import Resource
from api_demo.models.product import Product
from api_demo.api.schemas.product import ProductSchema
from .products_ns import ns, envelope_list, product_model, product_input, error_model

product_schema = ProductSchema()
products_schema = ProductSchema(many=True)


@ns.route("")
class ProductList(Resource):
    @ns.doc("list_products", params={"page": "页码", "per_page": "每页条数"})
    @ns.marshal_with(envelope_list, code=200)
    def get(self):
        """商品分页列表"""
        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)
        return {
            "code": 0, "message": "ok",
            "data": products_schema.dump(pagination.items),
            "pagination": {"page": pagination.page, "per_page": pagination.per_page,
                           "total": pagination.total, "pages": pagination.pages},
        }

    @ns.doc("create_product", security="Bearer")
    @ns.expect(product_input, validate=True)
    @ns.response(401, "未认证", error_model)
    @ns.response(422, "校验失败", error_model)
    def post(self):
        """创建商品（需登录）"""
        # 业务逻辑同 ch13 ...
        return {"code": 0, "message": "created", "data": {}}, 201


@ns.route("/<int:product_id>")
@ns.param("product_id", "商品 ID")
class ProductItem(Resource):
    @ns.marshal_with(ns.model("ProductDetailEnvelope", {
        "code": fields.Integer, "message": fields.String,
        "data": fields.Nested(product_model),
    }))
    @ns.response(404, "未找到", error_model)
    def get(self, product_id):
        product = Product.query.get_or_404(product_id)
        return {"code": 0, "message": "ok", "data": product_schema.dump(product)}