用 Flask 写 REST API 该怎么组织?MethodView 和版本化怎么做?
简化版
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 或游标),并把 total、has_next 放进响应。几个必须做对的细节:PUT 是整体替换、PATCH 是局部更新(PATCH 要能区分「没传字段」和「传了 null」)、201 要带 Location 头、204 不能有响应体、401 是「你是谁」403 是「不许」、写接口要考虑幂等(用 Idempotency-Key 或业务唯一键)。如果不想自己搭这套,可以用 flask-smorest(基于 marshmallow,自动生成 OpenAPI 文档)——但新项目如果 API 是主体,直接用 FastAPI 通常更省事。核心记忆:MethodView 组织资源;蓝图做版本;marshmallow/pydantic 管进出;errorhandler 统一错误格式。
详细版
REST 接口设计速查:
| 操作 | 方法 + 路径 | 成功状态码 | 说明 |
|---|---|---|---|
| 列表 | GET /articles | 200 | 分页 + 筛选 |
| 详情 | GET /articles/1 | 200 | 不存在 404 |
| 创建 | POST /articles | 201 + Location | 幂等性要考虑 |
| 整体更新 | PUT /articles/1 | 200 / 204 | 未传字段视为清空 |
| 局部更新 | PATCH /articles/1 | 200 | 只改传了的字段 |
| 删除 | DELETE /articles/1 | 204(无 body) | 幂等:重复删也返回 204 |
| 子资源 | GET /articles/1/comments | 200 | 层级不超过两层 |
# ① ★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。③PUT和PATCH的语义差别要落实到代码: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、分页,子类只声明 model 和 schema 的差异。注册风格上推荐把列表和详情拆成两个类(ArticleListAPI 和 ArticleItemAPI),比在 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 结构才需要升版本;而新增可选字段、新增接口、新增可选参数是向后兼容的(客户端会忽略不认识的字段)——经验是能兼容就别升版本,每个版本都是长期的维护负担。下线老版本要走完整流程:公告、响应加 Deprecation 和 Sunset 头(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_id、is_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基类把「鉴权 → 取对象 → 校验权限 → 分页 → 序列化」这套流程写一次,各资源子类只声明model和schema的差异,这是函数视图靠装饰器堆叠很难达到的。另外还有两个实际好处:类属性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标记id、created_at、author这类只应输出的字段,用load_only=True标记password这类只应输入的字段。 - 误区:注册一个
@app.errorhandler(Exception)就能统一所有错误响应了。 会把 404、405、400 全部变成 500。因为HTTPException(NotFound、MethodNotAllowed、BadRequest等)也是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供用户报障)。 - 误区:
PUT和PATCH差不多,随便用哪个都行。 语义完全不同,而且会导致真实的数据丢失。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.decode时algorithms参数必须写死白名单(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: true和Sunset: <日期>头(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 回退到 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 的 Number 只有 53 位精度)、空列表返回 [] 而不是 null。最后说句诚实话:纯 API 的新项目用 FastAPI 通常更省事,Flask + smorest/marshmallow 更适合给已有 Flask 应用加 API。