← 返回题目列表

用 Flask 写 REST API 该怎么组织?MethodView 和版本化怎么做?

中等 第 22 / 27 题 更新于 2026/08/02
FlaskREST APIMethodView版本化DRF对比

简化版

Flask 本身没有 API 框架的架子,所以「怎么组织」全靠约定,一套成熟的做法包含五层:① 路由用 MethodView 类视图——把同一资源的 get/post/put/patch/delete 组织在一个类里,Flask 按 request.method 自动分派,还能用 decorators = [login_required] 给所有方法统一加装饰器;② 版本化用蓝图 + url_prefix/api/v1),因为URL 路径版本最直观、最好调试、也最容易在网关层做分流,比 Header 版本和内容协商实用得多;③ 输入校验和输出序列化交给 marshmallow 或 pydantic——手写一堆 if not data.get("x"): abort(400) 既重复又无法生成文档;④ 错误响应必须统一格式——注册 @app.errorhandler(HTTPException) 一行就让所有 4xx/5xx 返回 JSON 而不是 HTML 页面,再定义自己的业务异常基类;⑤ 列表接口统一分页约定page/per_page 或游标),并把 totalhas_next 放进响应。几个必须做对的细节PUT 是整体替换、PATCH 是局部更新PATCH 要能区分「没传字段」和「传了 null」)、201 要带 Location204 不能有响应体401 是「你是谁」403 是「不许」写接口要考虑幂等(用 Idempotency-Key 或业务唯一键)。如果不想自己搭这套,可以用 flask-smorest(基于 marshmallow,自动生成 OpenAPI 文档)——但新项目如果 API 是主体,直接用 FastAPI 通常更省事。核心记忆:MethodView 组织资源蓝图做版本marshmallow/pydantic 管进出errorhandler 统一错误格式

详细版

REST 接口设计速查

操作方法 + 路径成功状态码说明
列表GET /articles200分页 + 筛选
详情GET /articles/1200不存在 404
创建POST /articles201 + Location幂等性要考虑
整体更新PUT /articles/1200 / 204未传字段视为清空
局部更新PATCH /articles/1200只改传了的字段
删除DELETE /articles/1204(无 body)幂等:重复删也返回 204
子资源GET /articles/1/comments200层级不超过两层
# ① ★MethodView 组织资源★
from flask import Blueprint, request, abort
from flask.views import MethodView

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

class ArticleAPI(MethodView):
    decorators = [login_required]              # ★★所有方法都加★★
    init_every_request = False                 # ★2.2+:复用实例,性能更好★

    def get(self, article_id):
        if article_id is None:                 # ★★同一个类处理列表和详情★★
            args = ListQuerySchema().load(request.args)
            q = Article.query.filter_by(status=args.get("status"))
            page = q.paginate(page=args["page"], per_page=args["per_page"])
            return {
                "items": ArticleSchema(many=True).dump(page.items),
                "meta": {"page": page.page, "per_page": page.per_page,
                         "total": page.total, "pages": page.pages,
                         "has_next": page.has_next},
            }
        art = db.get_or_404(Article, article_id)
        return ArticleSchema().dump(art)

    def post(self):
        data = ArticleCreateSchema().load(request.get_json(silent=True) or {})
        art = Article(**data, author=current_user)
        db.session.add(art); db.session.commit()
        return ArticleSchema().dump(art), 201, {                # ★★201 + Location★★
            "Location": url_for("api_v1.article_api", article_id=art.id)
        }

    def put(self, article_id):                 # ★整体替换★
        art = db.get_or_404(Article, article_id)
        check_owner(art)
        data = ArticleSchema().load(request.get_json())          # ★所有字段必填★
        for k, v in data.items(): setattr(art, k, v)
        db.session.commit()
        return ArticleSchema().dump(art)

    def patch(self, article_id):               # ★局部更新★
        art = db.get_or_404(Article, article_id)
        check_owner(art)
        data = ArticleSchema(partial=True).load(request.get_json())  # ★★partial★★
        for k, v in data.items(): setattr(art, k, v)
        db.session.commit()
        return ArticleSchema().dump(art)

    def delete(self, article_id):
        art = Article.query.get(article_id)
        if art:                                 # ★★幂等:不存在也返回 204★★
            check_owner(art); db.session.delete(art); db.session.commit()
        return "", 204                          # ★★204 不能有 body★★

# ★注册:一个类挂两条路由★
view = ArticleAPI.as_view("article_api")
api_v1.add_url_rule("/articles", defaults={"article_id": None},
                    view_func=view, methods=["GET"])
api_v1.add_url_rule("/articles", view_func=view, methods=["POST"])
api_v1.add_url_rule("/articles/<int:article_id>", view_func=view,
                    methods=["GET", "PUT", "PATCH", "DELETE"])

# ② ★统一错误格式★
from werkzeug.exceptions import HTTPException

class BizError(Exception):
    code, status, message = "biz_error", 400, "业务错误"
    def __init__(self, message=None, code=None, **extra):
        self.message = message or self.message
        self.code = code or self.code
        self.extra = extra

class InsufficientBalance(BizError):
    code, status, message = "insufficient_balance", 409, "余额不足"

def register_errors(app):
    @app.errorhandler(HTTPException)            # ★★所有 4xx/5xx 转 JSON★★
    def on_http(e):
        return {"error": {"code": e.name.lower().replace(" ", "_"),
                          "message": e.description}}, e.code

    @app.errorhandler(ValidationError)          # marshmallow
    def on_validation(e):
        return {"error": {"code": "validation_error",
                          "message": "参数校验失败",
                          "fields": e.messages}}, 422

    @app.errorhandler(BizError)
    def on_biz(e):
        return {"error": {"code": e.code, "message": e.message, **e.extra}}, e.status

    @app.errorhandler(Exception)                # ★兜底★
    def on_unexpected(e):
        if isinstance(e, HTTPException): return e      # ★★别把 404 变成 500★★
        app.logger.exception("unhandled")
        return {"error": {"code": "internal_error",
                          "message": "服务器内部错误",
                          "request_id": g.get("request_id")}}, 500

⚠️ 三个必须记住的点:① MethodView 的价值在「组织」和「复用」,不是语法糖。它让一个资源的所有操作聚在一个类里(不用在文件里翻找散落的 article_get/article_put),可以定义 BaseAPI 基类统一处理鉴权、序列化、分页,用类属性 decorators = [...] 批量加装饰器。Flask 2.2+ 还支持 init_every_request = False 让整个应用共享一个视图实例(不再每请求实例化,性能更好,但实例上就不能存请求相关状态了)。② 错误响应必须统一,而且要注册 HTTPException 的处理器。Flask 默认的 404/405/500 返回的是 HTML 错误页,API 客户端拿到会直接解析失败——一行 @app.errorhandler(HTTPException) 就能全部转成 JSON。注册 @app.errorhandler(Exception) 兜底时必须先判断 isinstance(e, HTTPException) 并原样返回,否则你的 404 会变成 500。③ PUTPATCH 的语义差别要落实到代码PUT整体替换(未传的字段应该被重置为默认值或报错),PATCH局部更新(只改传了的字段)。marshmallow 用 partial=True 表达 PATCH。而且 PATCH 要能区分「字段没传」和「传了 null」——前者是「不修改」,后者是「清空」,用 "field" in data 判断而不是 data.get("field") is None

完整版教学

一、MethodView 与视图组织

★ 三种组织方式的演进:
  ① ★函数视图(小项目够用)★
     @bp.get("/articles")          def list_articles(): ...
     @bp.post("/articles")         def create_article(): ...
     @bp.get("/articles/<int:id>") def get_article(id): ...
     ✗ ★同一资源的逻辑散落★、★公共处理(鉴权/序列化)要重复写★

  ② ★MethodView(★推荐★)★
     class ArticleAPI(MethodView):
         def get(self, id=None): ...
         def post(self): ...
     ✓ ★聚合、可继承、decorators 批量加★

  ③ ★flask-smorest / flask-restx(框架级)★
     ✓ ★自动 OpenAPI 文档★、参数解析、响应序列化
     ✗ 多一层抽象和依赖

★ ★MethodView 的分派机制★:
  as_view("name") 返回一个函数:
    def view(*args, **kwargs):
        self = ArticleAPI()                      # ★默认每请求创建实例★
        meth = getattr(self, request.method.lower(), None)
        if meth is None and request.method == "HEAD":
            meth = getattr(self, "get", None)    # ★HEAD 回退到 get★
        return current_app.ensure_sync(meth)(*args, **kwargs)
  ★ ★methods 会自动从类里定义的方法推导★
    (定义了 get/post → methods = ["GET", "POST", "HEAD", "OPTIONS"])

★ ★init_every_request(2.2+)★:
  class ArticleAPI(MethodView):
      init_every_request = False      # ★★整个应用共享一个实例★★
  ★ 好处:★少一次对象创建★(高 QPS 下有意义)
  ★ 代价:★★不能在 self 上存请求状态★★(线程不安全!)
    ✗ def get(self): self.user = current_user   # ★★会串数据★★
    ✓ 用 g 存请求级状态

★ ★基类复用(★MethodView 最大的价值★)★:
  class BaseAPI(MethodView):
      decorators = [login_required]
      model = None
      schema = None

      def get_object(self, oid):
          obj = db.get_or_404(self.model, oid)
          self.check_permission(obj)
          return obj

      def check_permission(self, obj):
          if getattr(obj, "user_id", None) != current_user.id:
              abort(403)

      def paginated(self, query):
          args = PageSchema().load(request.args)
          p = query.paginate(page=args["page"], per_page=args["per_page"],
                             error_out=False)
          return {"items": self.schema(many=True).dump(p.items),
                  "meta": {"page": p.page, "total": p.total,
                           "pages": p.pages, "has_next": p.has_next}}

  class ArticleAPI(BaseAPI):
      model, schema = Article, ArticleSchema     # ★子类只声明差异★

★ ★注册的两种风格★:
  # 风格一:手动 add_url_rule(灵活)
  view = ArticleAPI.as_view("article_api")
  bp.add_url_rule("/articles", defaults={"aid": None}, view_func=view,
                  methods=["GET"])
  bp.add_url_rule("/articles", view_func=view, methods=["POST"])
  bp.add_url_rule("/articles/<int:aid>", view_func=view,
                  methods=["GET", "PUT", "PATCH", "DELETE"])

  # 风格二:★拆成两个类(★更清晰★)★
  class ArticleListAPI(MethodView):     # /articles
      def get(self): ...                # 列表
      def post(self): ...               # 创建
  class ArticleItemAPI(MethodView):     # /articles/<id>
      def get(self, aid): ...
      def put(self, aid): ...
      def delete(self, aid): ...
  ★ ★推荐风格二★:不用在 get 里 if id is None,职责更单一

★ ★注册辅助函数(★大项目必备★)★:
  def register_api(bp, view_cls, name, url, pk="id", pk_type="int"):
      view = view_cls.as_view(name)
      bp.add_url_rule(url, defaults={pk: None}, view_func=view, methods=["GET"])
      bp.add_url_rule(url, view_func=view, methods=["POST"])
      bp.add_url_rule(f"{url}/<{pk_type}:{pk}>", view_func=view,
                      methods=["GET", "PUT", "PATCH", "DELETE"])
  register_api(api_v1, ArticleAPI, "article_api", "/articles", pk="article_id")

视图组织的演进是「函数视图 → MethodView → 框架级方案」MethodView 的分派机制很简单:as_view() 返回的函数按 request.method.lower() 取对应方法(HEAD 会回退到 get),methods 从类里定义的方法自动推导Flask 2.2+ 的 init_every_request = False 能让整个应用共享一个实例(省一次对象创建),但代价是不能在 self 上存请求状态——那会线程不安全MethodView 最大的价值是基类复用:定义一个 BaseAPI 处理鉴权、get_object、分页,子类只声明 modelschema 的差异。注册风格上推荐把列表和详情拆成两个类ArticleListAPIArticleItemAPI),比在 get() 里写 if id is None 职责更单一;大项目可以写个 register_api 辅助函数统一注册。

二、版本化策略

★ 四种版本化方式对比:
  ┌────────────────────┬────────────────────────────────────────┐
  │ ★① URL 路径★        │ ★/api/v1/articles★                      │
  │                     │ ✓ ★最直观、好调试、浏览器能直接访问★    │
  │                     │ ✓ ★网关/Nginx 能按路径分流★             │
  │                     │ ✓ ★缓存友好(不同版本不同 URL)★        │
  │                     │ ✗ 不够"RESTful 纯粹"(URI 应标识资源)  │
  ├────────────────────┼────────────────────────────────────────┤
  │ ② 请求头            │ X-API-Version: 2                        │
  │                     │ ✓ URL 干净                              │
  │                     │ ✗ ★浏览器测不了、抓包才看得到、缓存麻烦★│
  ├────────────────────┼────────────────────────────────────────┤
  │ ③ 内容协商          │ Accept: application/vnd.x.v2+json       │
  │                     │ ✓ 最"标准"                              │
  │                     │ ✗ ★最难用、客户端支持差★                │
  ├────────────────────┼────────────────────────────────────────┤
  │ ④ 查询参数          │ /api/articles?version=2                 │
  │                     │ ✗ ★容易被忽略、和业务参数混淆★          │
  └────────────────────┴────────────────────────────────────────┘
  ★ ★实践首选 URL 路径版本★(GitHub、Stripe 早期、绝大多数国内 API 都这么做)

★ ★Flask 里的实现★:
  # app/api/v1/__init__.py
  api_v1 = Blueprint("api_v1", __name__, url_prefix="/api/v1")
  from . import articles, users          # ★导入以注册路由★

  # app/api/v2/__init__.py
  api_v2 = Blueprint("api_v2", __name__, url_prefix="/api/v2")

  # create_app
  app.register_blueprint(api_v1)
  app.register_blueprint(api_v2)

★ ★v2 复用 v1 的代码(★避免全量复制★)★:
  # v2/articles.py
  from ..v1.articles import ArticleAPI as ArticleAPIv1

  class ArticleAPI(ArticleAPIv1):        # ★★继承 v1,只改差异★★
      schema = ArticleSchemaV2           # 新的序列化格式
      def get(self, aid=None):
          data = super().get(aid)
          data["new_field"] = ...        # ★只覆盖变化的部分★
          return data
  ★ 原则:★v2 只写"和 v1 不同的地方"★

★ ★什么时候才该升版本(★重要判断★)★:
  ★需要新版本(breaking change)★:
    ✗ 删除字段 / 改字段名 / 改字段类型
    ✗ 改变字段语义(status 从字符串变数字)
    ✗ 新增必填参数
    ✗ 改变默认行为(默认排序、默认分页大小)
    ✗ 改 URL 结构
  ★不需要新版本(向后兼容)★:
    ✓ ★新增可选字段★(客户端会忽略不认识的字段)
    ✓ ★新增接口★
    ✓ ★新增可选参数★
    ✓ 修 bug(除非有人依赖这个 bug)
  ★ ★经验:能兼容就别升版本——每个版本都是长期的维护负担★

★ ★版本下线流程★:
  ① ★公告 + 文档标记 deprecated★
  ② ★响应加头★:
     Deprecation: true
     Sunset: Sat, 31 Dec 2026 23:59:59 GMT       # ★RFC 8594★
     Link: <https://docs/x/v2>; rel="successor-version"
  ③ ★监控 v1 的调用量和调用方★(谁还在用)
  ④ ★逐步降级★:先限流 → 再间歇性返回 410 → 最后关闭
  ★ ★别直接关★——你不知道谁的定时任务还在调

★ ★字段级的兼容技巧★:
  # 改名时:★两个字段都返回一段时间★
  {"user_name": "x", "username": "x"}    # 新旧并存
  # 加必填参数时:★给一个默认值保持兼容★

版本化实践首选 URL 路径版本/api/v1/)——最直观、好调试、浏览器能直接访问、网关能按路径分流、缓存友好;Header 版本和内容协商虽然「更 RESTful」,但浏览器测不了、抓包才看得到、缓存麻烦。Flask 里就是每个版本一个蓝图 + url_prefix,而 v2 应该继承 v1 的类只覆盖差异,避免全量复制。「什么时候才该升版本」是个重要判断删字段、改字段名/类型/语义、新增必填参数、改默认行为、改 URL 结构才需要升版本;而新增可选字段、新增接口、新增可选参数是向后兼容的(客户端会忽略不认识的字段)——经验是能兼容就别升版本,每个版本都是长期的维护负担。下线老版本要走完整流程:公告、响应加 DeprecationSunset 头(RFC 8594)、监控调用方、逐步降级——别直接关,你不知道谁的定时任务还在调

三、序列化与校验

★ 手写校验的问题(★为什么要用库★):
  data = request.get_json(silent=True) or {}
  title = data.get("title")
  if not title: abort(400, "title 必填")
  if len(title) > 200: abort(400, "title 太长")
  status = data.get("status", "draft")
  if status not in ("draft", "published"): abort(400, "status 非法")
  ...
  ★ 问题:★重复、错误信息不统一、无法生成文档、改字段要改多处★

★ ★marshmallow(Flask 生态主流)★:
  from marshmallow import Schema, fields, validate, validates_schema

  class ArticleSchema(Schema):
      id = fields.Int(dump_only=True)                  # ★★只出不进★★
      title = fields.Str(required=True,
                         validate=validate.Length(1, 200))
      content = fields.Str(required=True)
      status = fields.Str(load_default="draft",
                          validate=validate.OneOf(["draft", "published"]))
      author = fields.Nested("UserSchema", dump_only=True)   # ★嵌套★
      created = fields.DateTime(dump_only=True, format="iso")
      password = fields.Str(load_only=True)            # ★★只进不出(不返回)★★

      @validates_schema
      def check(self, data, **kw):                     # ★跨字段校验★
          if data.get("status") == "published" and not data.get("content"):
              raise ValidationError("发布时内容不能为空", "content")

  # 使用
  data = ArticleSchema().load(request.get_json())      # ★校验 + 反序列化★
  out = ArticleSchema().dump(article)                  # ★序列化★
  out = ArticleSchema(many=True).dump(articles)        # 列表
  data = ArticleSchema(partial=True).load(...)         # ★★PATCH 用★★
  out = ArticleSchema(only=("id", "title")).dump(a)    # ★字段裁剪★
  out = ArticleSchema(exclude=("content",)).dump(a)

★ ★dump_only / load_only 的意义★:
  dump_only=True  → ★输出时包含,输入时忽略★(id、created_at)
    ★ 防止用户通过请求体篡改 id 或创建时间
  load_only=True  → ★输入时接受,输出时排除★(password)
    ★ 防止密码被序列化返回

★ ★★Mass Assignment 漏洞(★必须防★)★★:
  ✗ art = Article(**request.get_json())           # ★★危险★★
    攻击:{"title": "x", "author_id": 999, "is_featured": true}
    → ★用户能设置任何字段★
  ✓ 用 Schema 白名单(★未声明的字段会被拒绝或忽略★)
    class Meta: unknown = RAISE     # ★★未知字段直接报错(推荐)★★
    # 或 EXCLUDE(忽略)/ INCLUDE(危险,别用)

★ ★pydantic 方案★:
  from pydantic import BaseModel, Field, field_validator
  class ArticleCreate(BaseModel):
      title: str = Field(min_length=1, max_length=200)
      content: str
      status: Literal["draft", "published"] = "draft"
      model_config = {"extra": "forbid"}        # ★★拒绝未知字段★★

  class ArticleOut(BaseModel):
      id: int
      title: str
      created: datetime
      model_config = {"from_attributes": True}  # ★从 ORM 对象构造★

  # 使用
  try:
      data = ArticleCreate(**(request.get_json(silent=True) or {}))
  except ValidationError as e:
      return {"error": {"code": "validation_error", "fields": e.errors()}}, 422
  return ArticleOut.model_validate(art).model_dump(mode="json")

★ ★两者选择★:
  ┌──────────────┬────────────────────────────────────┐
  │ ★marshmallow★ │ ★Flask 生态成熟★、和 SQLAlchemy 集成 │
  │              │ 好(marshmallow-sqlalchemy 自动生成)│
  │              │ ★flask-smorest 基于它自动出 OpenAPI★ │
  │ ★pydantic★    │ ★类型注解、IDE 友好、v2 性能好★      │
  │              │ ★从 FastAPI 过来零成本★              │
  └──────────────┴────────────────────────────────────┘

★ ★序列化的 N+1 陷阱★:
  ArticleSchema(many=True).dump(articles)    # ★每个 article 访问 .author★
  → ★N+1!★
  ✓ 查询时预加载:
    Article.query.options(joinedload(Article.author)).all()

手写校验的问题是重复、错误信息不统一、无法生成文档marshmallow 的两个关键选项dump_only=True(只出不进,防止用户通过请求体篡改 id 或创建时间)和 load_only=True(只进不出,防止密码被序列化返回)。必须防的是 Mass Assignment 漏洞——Article(**request.get_json()) 让用户能设置任何字段(包括 author_idis_featured),用 Schema 白名单并设 unknown = RAISE 是正解。marshmallow 和 pydantic 的选择:Flask 生态成熟、和 SQLAlchemy 集成好、flask-smorest 基于它出 OpenAPI 的选前者;类型注解友好、从 FastAPI 过来的选后者。最后一个易忽略的点:序列化嵌套字段会触发 N+1,查询时要预加载。

四、分页、筛选与响应约定

★ 分页的两种方案:
  ┌────────────────┬──────────────────────────────────────┐
  │ ★偏移分页★      │ ?page=2&per_page=20                   │
  │                 │ ✓ ★能跳页、能显示总数★                 │
  │                 │ ✗ ★深分页慢、数据变动时会重复/漏★      │
  │ ★游标分页★      │ ?cursor=xxx&limit=20                  │
  │                 │ ✓ ★恒定性能、不重不漏★                 │
  │                 │ ✗ ★不能跳页、不能显示总页数★           │
  └────────────────┴──────────────────────────────────────┘
  ★ ★后台管理用偏移、移动端信息流用游标★

★ ★统一的响应结构(★团队要先约定★)★:
  # 方案 A:★数据和元信息分开(推荐)★
  {
    "items": [...],
    "meta": {"page": 2, "per_page": 20, "total": 153,
             "pages": 8, "has_next": true, "has_prev": true}
  }
  # 方案 B:★JSON:API 风格★
  {"data": [...], "links": {"next": "...", "prev": "..."},
   "meta": {"total": 153}}
  # 方案 C:★分页信息放响应头(★HTML 内容时常用★)★
  X-Total-Count: 153
  Link: <...?page=3>; rel="next", <...?page=8>; rel="last"

  ★ ★关键是全站统一★——最怕的是有的接口返回 items、
    有的返回 data、有的直接返回数组

★ ★Flask-SQLAlchemy 的分页★:
  p = query.paginate(page=1, per_page=20,
                     error_out=False,        # ★★页码超范围返回空而不是 404★★
                     max_per_page=100)       # ★★防止 per_page=99999★★
  p.items / p.total / p.pages / p.has_next / p.next_num
  ★ ★per_page 必须有上限★(否则一个请求能拖垮数据库)
  ★ ★total 需要一次 COUNT★——大表上很贵,考虑估算或不返回

★ ★筛选与排序的约定★:
  GET /articles?status=published&author_id=3&created_after=2026-01-01
                &sort=-created&fields=id,title
  ★ 实现要点:
    ① ★白名单★:可筛选/可排序的字段要枚举,★不能直接把参数塞进 filter★
       ✗ query.filter_by(**request.args)     # ★★危险★★
       ✓ ALLOWED = {"status", "author_id"}
    ② ★排序前缀 - 表示倒序★:sort=-created
       field = sort.lstrip("-"); desc = sort.startswith("-")
       if field not in SORTABLE: abort(400)
    ③ ★fields 稀疏字段集★:Schema(only=fields.split(","))
    ④ ★范围查询命名统一★:created_after / created_before 或 created[gte]

★ ★响应约定的其他要点★:
  □ ★时间统一 ISO 8601 + UTC★:"2026-08-02T10:00:00Z"
  □ ★金额用字符串★(避免浮点精度):"99.90"
  □ ★大整数 ID 用字符串★(JS 的 Number 只有 53 位精度)
  □ ★布尔就用 true/false★(不要 0/1 或 "yes")
  □ ★空列表返回 []★(不要 null)
  □ ★字段命名风格统一★(snake_case 或 camelCase 全站一致)
  □ ★不返回 null 字段 vs 返回 null★——★选一种并写进文档★

★ ★HATEOAS 要不要做★:
  {"id": 1, "_links": {"self": "/articles/1",
                       "comments": "/articles/1/comments"}}
  ★ 理论上是 REST 成熟度模型的最高级
  ★ ★现实中极少有客户端真的按链接导航★
  → ★大多数项目不做,或只在少数场景(如分页链接)用★

分页有两套方案:后台管理用偏移分页(能跳页、显示总数),移动端信息流用游标分页(恒定性能、不重不漏)。响应结构的关键是全站统一——最怕有的接口返回 items、有的返回 data、有的直接返回数组。用 Flask-SQLAlchemy 的 paginate两个参数必须设error_out=False(页码超范围返回空而不是 404)和 max_per_page(防止 per_page=99999 拖垮数据库)。筛选和排序的实现要点是白名单——绝对不能 query.filter_by(**request.args),可筛选和可排序的字段必须枚举。响应约定里几条容易被忽略的:时间统一 ISO 8601 + UTC金额用字符串避免浮点精度大整数 ID 用字符串(JS 的 Number 只有 53 位精度)、空列表返回 [] 而不是 null。至于 HATEOAS——理论上是 REST 成熟度的最高级,但现实中极少有客户端真的按链接导航,大多数项目不做。

五、认证、幂等与限流

★ 认证方案对比:
  ┌──────────────┬────────────────────────────────────────┐
  │ ★Session★     │ 浏览器友好、★可即时失效★                │
  │              │ ✗ ★需要 CSRF 防护★、跨域麻烦、有状态     │
  │ ★JWT★         │ ★无状态、跨服务方便★                    │
  │              │ ✗ ★★无法主动失效★★(除非维护黑名单)     │
  │              │ ✗ ★体积大(每次请求都带)★              │
  │ ★API Key★     │ 服务端到服务端、简单                    │
  │              │ ✗ 无法表达用户身份、★泄露即完全暴露★    │
  │ ★OAuth2★      │ ★第三方授权的标准★                      │
  └──────────────┴────────────────────────────────────────┘

★ ★JWT 的正确用法(★坑很多★)★:
  import jwt
  token = jwt.encode({"sub": str(user.id),
                      "exp": now + timedelta(minutes=15),   # ★★短期★★
                      "iat": now, "jti": uuid4().hex},
                     SECRET, algorithm="HS256")
  # 验证
  try:
      payload = jwt.decode(token, SECRET,
                           algorithms=["HS256"])   # ★★必须指定算法白名单★★
  except jwt.ExpiredSignatureError: abort(401, "token 已过期")
  except jwt.InvalidTokenError: abort(401, "token 无效")

  ★ ★三个必须注意★:
    ① ★algorithms 必须写死★——否则 ★alg: none 攻击★
    ② ★access token 要短(15 分钟)+ refresh token 长期★
       → ★因为 JWT 无法主动失效★,短期能限制损失
    ③ ★不要在 payload 里放敏感信息★——★JWT 只是签名,不是加密★
       (base64 解开就能看到内容)
  ★ ★需要"立即登出"就必须维护黑名单(Redis)★
    → 那就已经有状态了 → ★不如直接用 session★

★ ★★幂等性(写接口的核心问题)★★:
  场景:用户点了两次"提交订单",或客户端超时重试
  ✗ 不处理 → ★重复下单★

  ✓ 方案一:★Idempotency-Key 头(★Stripe 的做法★)★
     @api.post("/orders")
     def create_order():
         key = request.headers.get("Idempotency-Key")
         if key:
             cached = redis.get(f"idem:{key}")
             if cached: return json.loads(cached)      # ★★直接返回上次结果★★
         order = do_create()
         result = OrderSchema().dump(order)
         if key:
             redis.setex(f"idem:{key}", 86400, json.dumps(result))
         return result, 201
     ★ 要点:★还要处理"并发的同 key 请求"(加锁)★

  ✓ 方案二:★业务唯一键 + 数据库唯一约束★
     UniqueConstraint("user_id", "order_no")
     try:
         db.session.add(order); db.session.commit()
     except IntegrityError:
         db.session.rollback()
         return existing_order                          # ★★返回已有的★★

  ★ ★HTTP 方法的幂等性(★理论必考★)★:
    GET/PUT/DELETE/HEAD/OPTIONS → ★幂等★
    POST/PATCH → ★不幂等★
    ★ 注意:★幂等 ≠ 安全★
      GET/HEAD 是"安全的"(不改变状态)
      PUT/DELETE 幂等但不安全

★ ★限流★:
  from flask_limiter import Limiter
  limiter = Limiter(key_func=lambda: g.user_id or get_remote_address(),
                    storage_uri="redis://...")      # ★★必须共享存储★★
  @api.post("/login")
  @limiter.limit("5/minute")                        # ★登录接口重点限★
  def login(): ...
  ★ 429 响应要带:
    Retry-After: 60
    X-RateLimit-Limit / X-RateLimit-Remaining / X-RateLimit-Reset

★ ★CORS(前后端分离必配)★:
  CORS(app, resources={r"/api/*": {"origins": ["https://app.x.com"]}},
       supports_credentials=True,          # ★★带 Cookie 时★★
       allow_headers=["Content-Type", "Authorization", "Idempotency-Key"],
       expose_headers=["X-Total-Count"])   # ★★前端才能读到自定义头★★
  ★ ★supports_credentials=True 时 origins 不能是 *★

认证方案里 JWT 的坑最多algorithms 必须写死算法白名单(否则有 alg: none 攻击)、access token 要短(15 分钟)配 refresh token(因为 JWT 无法主动失效)、payload 里不能放敏感信息JWT 只是签名不是加密,base64 解开就能看);而且需要「立即登出」就必须维护黑名单,那就已经有状态了——不如直接用 session幂等性是写接口的核心问题:两种方案是 Idempotency-Key(Stripe 的做法,注意要处理并发同 key)和业务唯一键 + 数据库唯一约束(捕获 IntegrityError 返回已有记录)。理论上要记住:GET/PUT/DELETE 幂等,POST/PATCH 不幂等;幂等 ≠ 安全(PUT 幂等但会改状态)。CORS 有个容易漏的点:自定义响应头要加进 expose_headers 前端才读得到

六、文档、测试与选型

★ ★自动生成文档:flask-smorest★
  from flask_smorest import Api, Blueprint
  app.config["API_TITLE"] = "My API"
  app.config["API_VERSION"] = "v1"
  app.config["OPENAPI_VERSION"] = "3.0.2"
  api = Api(app)

  bp = Blueprint("articles", "articles", url_prefix="/api/v1/articles")

  @bp.route("/")
  class Articles(MethodView):
      @bp.arguments(ListQuerySchema, location="query")   # ★入参★
      @bp.response(200, ArticleSchema(many=True))        # ★出参★
      @bp.paginate()                                     # ★分页★
      def get(self, args, pagination): ...

      @bp.arguments(ArticleCreateSchema)
      @bp.response(201, ArticleSchema)
      @bp.alt_response(422, description="校验失败")
      def post(self, data): ...

  api.register_blueprint(bp)
  # → ★自动生成 /swagger-ui 和 openapi.json★
  ★ ★好处:文档和代码同源,不会脱节★

★ ★API 测试★:
  def test_create_article(client, auth_headers):
      r = client.post("/api/v1/articles",
                      json={"title": "x", "content": "y"},
                      headers=auth_headers)
      assert r.status_code == 201
      assert "Location" in r.headers                 # ★★检查规范★★
      assert r.json["title"] == "x"
      assert "password" not in r.json                # ★★不该泄露的字段★★

  def test_validation_error(client, auth_headers):
      r = client.post("/api/v1/articles", json={}, headers=auth_headers)
      assert r.status_code == 422
      assert r.json["error"]["code"] == "validation_error"
      assert "title" in r.json["error"]["fields"]    # ★错误结构也要测★

  def test_permission(client, other_user_headers, article):
      r = client.delete(f"/api/v1/articles/{article.id}",
                        headers=other_user_headers)
      assert r.status_code == 403                    # ★★权限是最该测的★★

  ★ ★该测的优先级★:
    ① ★权限★(未登录 401 / 无权 403 / 只能操作自己的)
    ② ★校验★(必填、边界、非法枚举)
    ③ ★状态码和响应结构★
    ④ ★N+1(assertNumQueries 式的查询计数)★
    ⑤ 业务规则

★ ★Flask vs FastAPI vs DRF(★选型必答★)★:
  ┌────────────┬────────────────────────────────────────────┐
  │ ★Flask★     │ ★灵活、生态成熟、什么都能接★                │
  │            │ ✗ ★API 的架子要自己搭(或用 smorest)★       │
  │            │ ✓ ★已有 Flask 项目要加 API 时的自然选择★     │
  ├────────────┼────────────────────────────────────────────┤
  │ ★FastAPI★   │ ★自动文档、类型驱动、async 原生★             │
  │            │ ✓ ★纯 API 新项目的首选★                      │
  │            │ ✗ 生态比 Flask 年轻、同步库要小心            │
  ├────────────┼────────────────────────────────────────────┤
  │ ★DRF★       │ ★ORM 深度集成、权限/序列化/视图集全套★       │
  │            │ ✓ ★已有 Django 项目★                        │
  │            │ ✗ 重、学习曲线陡                             │
  └────────────┴────────────────────────────────────────────┘
  ★ ★诚实的判断:如果 API 是项目主体且是新项目 → FastAPI 更省事★
    ★如果是给已有 Flask 应用加 API → Flask + smorest/marshmallow★

★ 检查清单:
  □ ★MethodView 组织资源,基类复用公共逻辑★
  □ ★URL 路径版本 + 蓝图★
  □ ★Schema 白名单(防 Mass Assignment,unknown=RAISE)★
  □ ★dump_only 保护 id,load_only 保护 password★
  □ ★errorhandler(HTTPException) 统一 JSON 错误★
  □ ★兜底 handler 里先判断 HTTPException★
  □ ★per_page 有上限★
  □ ★筛选排序字段白名单★
  □ ★201 带 Location、204 不带 body★
  □ ★写接口考虑幂等★
  □ ★JWT 指定 algorithms、短过期★
  □ ★限流用 Redis 存储★
  □ ★时间 ISO 8601、金额和大 ID 用字符串★
  □ ★优先测权限和校验★

★ 一句话总结:
  ★"Flask 写 API 靠约定:MethodView 组织资源、蓝图做版本、
    marshmallow/pydantic 管进出(白名单防 Mass Assignment)、
    errorhandler 统一错误格式、分页和筛选要有上限和白名单;
    写接口考虑幂等,JWT 记得锁死算法和缩短有效期。"★

flask-smorest 能让文档和代码同源(用装饰器声明入参出参,自动生成 Swagger UI 和 openapi.json),不会出现文档脱节。API 测试该测的优先级是:权限 > 校验 > 状态码和响应结构 > N+1 > 业务规则——权限是最该测的(未登录 401、无权 403、只能操作自己的数据)。选型上要诚实:如果 API 是项目主体且是新项目,FastAPI 通常更省事(自动文档、类型驱动、async 原生);Flask + smorest/marshmallow 更适合给已有 Flask 应用加 API

记忆钩子:「Flask 没有 API 框架的架子,★五层约定要自己搭★:★① MethodView 组织资源★(同一资源的 get/post/put/patch/delete 聚在一个类,★methods 从定义的方法自动推导、HEAD 回退到 get★,★decorators = [login_required] 批量加★,★2.2+ 的 init_every_request=False 复用实例但绝不能在 self 上存请求状态★)——★最大价值是基类复用★(BaseAPI 统一鉴权/get_object/分页,子类只声明 model 和 schema);★推荐把列表和详情拆成两个类★而不是在 get 里 if id is None。★② 版本化用『蓝图 + url_prefix』的 URL 路径版本★(最直观、好调试、★网关能按路径分流、缓存友好★,比 Header 版本和内容协商实用),★v2 继承 v1 只覆盖差异★;★判断要不要升版本:删字段/改名/改类型/改语义/加必填/改默认行为才需要,新增可选字段和新增接口是兼容的★——★能兼容就别升,每个版本都是长期维护负担★;下线要发 ★Deprecation 和 Sunset 头(RFC 8594)★ 并监控调用方。★③ marshmallow/pydantic 管进出★:★dump_only 保护 id 和 created(防用户篡改)、load_only 保护 password(防泄露)★,★必须用 Schema 白名单防 Mass Assignment(Article(**request.get_json()) 让用户能设任何字段),设 unknown=RAISE★;★PATCH 用 partial=True,且要用 ‘x’ in data 区分『没传』和『传了 null』★。★④ 统一错误格式★:★Flask 默认 404/500 返回 HTML,注册 errorhandler(HTTPException) 一行全转 JSON★;★兜底的 errorhandler(Exception) 里必须先 isinstance(e, HTTPException) 原样返回,否则 404 会变成 500★。★⑤ 分页筛选要有上限和白名单★:★paginate 必设 max_per_page 和 error_out=False★,★绝不能 query.filter_by(**request.args)★。状态码:★201 带 Location、204 不能有 body、401 是『你是谁』403 是『不许』、422 是格式对但校验失败、429 带 Retry-After★。★写接口必须考虑幂等★:Idempotency-Key(Stripe 做法)或★业务唯一键 + 数据库唯一约束捕获 IntegrityError★;★GET/PUT/DELETE 幂等,POST/PATCH 不幂等,幂等≠安全★。JWT 三个坑:★algorithms 必须写死(防 alg:none)、access token 要短(因为无法主动失效)、payload 只是签名不是加密★——★要能立即登出就得维护黑名单,那还不如用 session★。响应约定:★时间 ISO 8601+UTC、金额用字符串、大整数 ID 用字符串(JS 只有 53 位精度)、空列表返回 [] 不是 null★。★选型诚实话:纯 API 新项目 FastAPI 更省事,给已有 Flask 加 API 才用 smorest/marshmallow★。」

七、常见误区与追问

  • 误区:用 MethodView 只是为了少写几个装饰器,和函数视图没本质区别。 它真正的价值在组织和复用组织上,一个资源的所有操作聚在一个类里——不用在几百行的 views.py 里翻找散落的 article_list/article_create/article_update复用上,你可以定义 BaseAPI 基类把「鉴权 → 取对象 → 校验权限 → 分页 → 序列化」这套流程写一次,各资源子类只声明 modelschema 的差异,这是函数视图靠装饰器堆叠很难达到的。另外还有两个实际好处:类属性 decorators = [login_required, rate_limit] 会作用于所有方法Flask 2.2+ 的 init_every_request = False 能让整个应用共享一个视图实例(省掉每请求的对象创建),但代价是绝不能在 self 上存请求相关状态——那样多线程下会串数据,请求级状态一律用 g
  • 误区:Article(**request.get_json()) 写起来最简洁,字段多的时候很省事。 这是 Mass Assignment(批量赋值)漏洞的经典形态。攻击者只要在请求体里多塞几个字段——{"title": "x", "author_id": 999, "is_featured": true, "created_at": "2020-01-01"}——就能设置任何模型字段:把文章的作者改成别人、把自己的评论设成精选、伪造创建时间,甚至在用户模型上直接把 is_admin 设成 true。防御方式是白名单:用 marshmallow/pydantic 的 Schema 明确声明「哪些字段可以从请求体进来」,并且把未声明的字段显式拒绝(marshmallow 的 class Meta: unknown = RAISE、pydantic 的 model_config = {"extra": "forbid"})——设成「忽略」也可以,但报错比静默忽略更容易发现前端传错。同时用 dump_only=True 标记 idcreated_atauthor 这类只应输出的字段,用 load_only=True 标记 password 这类只应输入的字段。
  • 误区:注册一个 @app.errorhandler(Exception) 就能统一所有错误响应了。 会把 404、405、400 全部变成 500。因为 HTTPExceptionNotFoundMethodNotAllowedBadRequest 等)也是 Exception 的子类,会被这个兜底处理器捕获——于是用户访问一个不存在的 URL,拿到的是「服务器内部错误 500」,监控系统也会被大量假的 5xx 告警淹没。正确写法是在兜底处理器里先判断并原样返回if isinstance(e, HTTPException): return e。更完整的做法是注册三层处理器@app.errorhandler(HTTPException) 把所有标准 HTTP 错误转成 JSON(这一行就解决了「404 返回 HTML 页面」的问题)、@app.errorhandler(ValidationError) 处理参数校验(返回 422 并带上字段级错误)、@app.errorhandler(BizError) 处理自定义业务异常(带业务错误码),最后才是 Exception 兜底(只记日志、不返回堆栈,响应里带上 request_id 供用户报障)。
  • 误区:PUTPATCH 差不多,随便用哪个都行。 语义完全不同,而且会导致真实的数据丢失PUT 是「用请求体整体替换这个资源」——严格来说,请求体里没出现的字段应该被重置为默认值或清空PATCH 是「只修改请求体里出现的字段」,其余保持不变。如果你把 PUT 实现成了「只更新传了的字段」,那它其实是 PATCH;反过来如果客户端以为是 PATCH 语义而调了你「真正实现了整体替换」的 PUT,只传一个 title 就会把 content 清空。实现上:marshmallow 用 Schema(partial=True) 表达 PATCH(所有字段变可选)。还有一个更细但很关键的点:PATCH 必须能区分「字段没传」和「字段传了 null——前者是「不修改」,后者是「显式清空」;用 "field" in data 判断而不是 data.get("field") is None,否则用户永远无法把某个字段置空。
  • 误区:用 JWT 就能做到无状态认证,比 session 先进。 JWT 的「无状态」是优点也是最大的坑你无法主动让一个已签发的 token 失效。用户改了密码、账号被封禁、token 泄露了——在它自然过期之前,它依然完全有效。常见的补救是维护一个 Redis 黑名单(存已注销的 jti),但那样就重新变成有状态了,还不如一开始就用 session。所以正确的用法是:access token 设得很短(15 分钟左右)+ 长期的 refresh token(refresh token 存数据库、可以撤销),用短有效期来限制损失。另外两个必须注意的:jwt.decodealgorithms 参数必须写死白名单algorithms=["HS256"])——否则攻击者可以把 header 里的 alg 改成 none 或换成非对称算法混淆来伪造 token;② JWT 的 payload 只是 base64 编码,不是加密——任何人拿到 token 都能解开看内容,绝不能放手机号、身份证、内部 ID 之外的敏感信息。选型建议:同一个域下的 Web 应用用 session 更简单也更安全,JWT 适合真正需要跨服务、跨域、无共享存储的场景。
  • 追问:写接口的幂等性该怎么保证? 问题的根源是网络的不确定性:客户端发了创建订单的请求,超时了——但它不知道服务端到底有没有执行成功,重试就可能重复下单。两种主流方案。Idempotency-Key 请求头(Stripe 的做法):客户端为每次「逻辑操作」生成一个唯一 key(通常是 UUID)随请求带上,服务端第一次执行后把结果缓存到 Redis(保留 24 小时),后续同 key 的请求直接返回缓存的结果而不重新执行。要注意并发的同 key 请求——两个请求同时进来时缓存都还没写,需要用 Redis 的分布式锁或 SETNX 占位。② 业务唯一键 + 数据库唯一约束:给订单加 UniqueConstraint("user_id", "order_no"),插入时捕获 IntegrityError 并返回已存在的那条记录——这个方案更可靠,因为它把幂等性交给了数据库这个唯一的真相源,不依赖缓存。理论上还要记住 HTTP 方法的幂等性GET/PUT/DELETE/HEAD 是幂等的(多次执行效果同一次),POST/PATCH 不是;注意「幂等」和「安全」是两回事——GET/HEAD 是「安全的」(不改变状态),而 PUT/DELETE 幂等但会改状态。
  • 追问:API 版本什么时候必须升?怎么优雅地下线老版本? 需要升版本的是「破坏性变更」:删除字段、改字段名、改字段类型(status 从字符串变成数字)、改变字段语义(同名但含义变了,这是最危险的,因为客户端不会报错只会算错)、新增必填参数、改变默认行为(默认排序、默认分页大小)、改 URL 结构。不需要升版本的是向后兼容的变更:新增可选字段(客户端会忽略不认识的字段)、新增接口、新增可选参数、修 bug。经验是「能兼容就别升」——每个版本都要长期维护、测试要跑两遍、bug 要修两处。字段改名时有个实用技巧:新旧字段同时返回一段时间,等调用方都迁移完再删旧的。下线流程要走四步:① 提前公告并在文档标记 deprecated② 响应里加 Deprecation: trueSunset: <日期> 头(RFC 8594),让自动化工具能发现;③ 监控老版本的调用量和调用方(谁还在用、用了哪些接口);④ 逐步降级——先限流、再间歇性返回 410 Gone(让对方感知到)、最后才彻底关闭。绝对不要直接关——你不知道哪个客户的定时任务还在凌晨三点调用它。
  • 追问:既然 FastAPI 有自动文档和类型驱动,为什么还有人用 Flask 写 API? 三个现实原因。① 存量项目——已经有一个跑了几年的 Flask 应用(模板页面、后台、定时任务、一堆扩展),现在要加一组 API,没有理由为此引入第二个框架,用 flask-smorest 就能拿到 OpenAPI 文档和 marshmallow 校验,体验和 FastAPI 差距不大。② 生态和团队熟悉度——Flask 生态更成熟(Flask-Login、Flask-Admin、Flask-Migrate 这些开箱即用),团队已有的中间件、日志、监控、部署经验都能直接复用。③ 同步生态更省心——FastAPI 的优势建立在 async 之上,但如果你的数据库驱动、第三方 SDK 都是同步的,写在 async def 里反而会阻塞事件循环(要么小心地用 run_in_threadpool,要么全用 def 视图),收益就大打折扣了。反过来,诚实的判断是:如果是纯 API 的新项目、团队愿意用类型注解、依赖也支持异步,FastAPI 确实更省事——自动文档、依赖注入、pydantic 校验都是内置的,不用自己搭一遍上面说的五层约定。

八、加强记忆

Flask 没有 API 框架的架子,五层约定要自己搭MethodView 组织资源——同一资源的 get/post/put/patch/delete 聚在一个类里(methods 从定义的方法自动推导、HEAD 回退到 getdecorators = [login_required] 批量加装饰器2.2+ 的 init_every_request = False 可复用实例但绝不能在 self 上存请求状态)——它最大的价值是基类复用BaseAPI 统一鉴权、get_object、分页,子类只声明 modelschema);推荐把列表和详情拆成两个类而不是在 get() 里写 if id is None② 版本化用「蓝图 + url_prefix」的 URL 路径版本(最直观、好调试、网关能按路径分流、缓存友好,比 Header 版本和内容协商实用得多),v2 继承 v1 只覆盖差异判断要不要升版本:删字段、改名、改类型、改语义、加必填、改默认行为才需要,而新增可选字段和新增接口是向后兼容的——能兼容就别升,每个版本都是长期的维护负担;下线要发 DeprecationSunset 头(RFC 8594) 并监控调用方。③ marshmallow/pydantic 管进出dump_only 保护 idcreated(防用户篡改)、load_only 保护 password(防泄露)必须用 Schema 白名单防 Mass AssignmentArticle(**request.get_json()) 让用户能设置任何字段),并设 unknown = RAISEPATCHpartial=True,且要用 "x" in data 区分「没传」和「传了 null」④ 统一错误格式Flask 默认的 404/500 返回 HTML,注册 errorhandler(HTTPException) 一行就全转成 JSON兜底的 errorhandler(Exception) 里必须先 isinstance(e, HTTPException) 原样返回,否则 404 会变成 500⑤ 分页筛选要有上限和白名单paginate 必设 max_per_pageerror_out=False绝不能 query.filter_by(**request.args)。状态码要点:201 带 Location、204 不能有 body、401 是「你是谁」403 是「不许」、422 是格式对但校验失败、429 带 Retry-After写接口必须考虑幂等:用 Idempotency-Key(Stripe 做法)或业务唯一键 + 数据库唯一约束捕获 IntegrityError;理论上 GET/PUT/DELETE 幂等、POST/PATCH 不幂等,且幂等 ≠ 安全。JWT 有三个坑:algorithms 必须写死(防 alg: none)、access token 要短(因为无法主动失效)、payload 只是签名不是加密——要能立即登出就得维护黑名单,那还不如直接用 session。响应约定:时间用 ISO 8601 + UTC、金额用字符串、大整数 ID 用字符串(JS 的 Number 只有 53 位精度)、空列表返回 [] 而不是 null。最后说句诚实话:纯 API 的新项目用 FastAPI 通常更省事,Flask + smorest/marshmallow 更适合给已有 Flask 应用加 API