上一章我们用钩子把「每个请求都要做的事」抽了出来,这一章换个方向:把同一个资源上的多个动作收进一个类。学完你能用 MethodView 写出「一个类管一个资源」的 RESTful 接口,也能判断出哪些地方该用类视图、哪些地方函数视图反而更清爽——不为了「高级」而高级。
当某个资源的增删改查动作很多,函数视图会写出一堆零散函数、路由注册很多遍。类视图(MethodView)把同一资源的全部动作收进一个类,方法名直接对应 HTTP 方法。
from flask import Blueprint, request
from utils.response import success, fail
article_bp = Blueprint("article", __name__, url_prefix="/api/article")
# ① 列表
@article_bp.get("/list")
def article_list():
page = request.args.get("page", 1, type=int)
return success({"page": page, "list": []})
# ② 新增
@article_bp.post("/")
def article_create():
return success(None, "新增成功")
# ③ 详情
@article_bp.get("/detail/<int:aid>")
def article_detail(aid):
return success({"id": aid})
# ④ 修改
@article_bp.put("/<int:aid>")
def article_update(aid):
return success(None, "修改成功")
# ⑤ 删除
@article_bp.delete("/<int:aid>")
def article_delete(aid):
return success(None, "删除成功")
from flask import Flask
from flask.views import MethodView
from utils.response import success
app = Flask(__name__)
class HealthView(MethodView):
"""健康检查:只有查,没有增删改,所以只实现 get()"""
def get(self):
return success({"service": "hiflask", "status": "up"})
# as_view 把类「转换」成视图函数,再交给 add_url_rule 注册路由
app.add_url_rule("/api/health", view_func=HealthView.as_view("health"))
路由表要的是可调用的函数,类本身不是函数。as_view() 把类转成一个视图函数并指定端点名,之后才能用 add_url_rule 注册。
from flask import Flask, url_for
from flask.views import MethodView
app = Flask(__name__)
class ArticleDetailView(MethodView):
def get(self, aid):
return {"code": 200, "msg": "success", "data": {"id": aid}}
# 关键:as_view() 返回的是「视图函数」,add_url_rule 只认函数
app.add_url_rule("/api/article/detail/<int:aid>",
view_func=ArticleDetailView.as_view("article_detail"))
with app.test_request_context():
# url_for 的第一个参数就是 as_view 里写的名字
print(url_for("article_detail", aid=3))
注册类视图后,Flask 会按请求的实际 HTTP 方法自动调用类里同名的方法(GET→get、POST→post…)。你不再手写「if request.method == …」分支。
# 先看这个类视图支持哪些方法
$ flask --app app routes
Endpoint Methods Rule
-------------- ----------------- -----------
health GET, HEAD, OPTIONS /api/health
# 类里没写 delete(),DELETE 请求会被 Werkzeug 直接挡掉
$ curl -i -X DELETE http://127.0.0.1:5000/api/health
一个资源通常拆成两个类:列表类挂在 /articles 负责 GET 列表与 POST 新建,详情类挂在 /articles/
from flask import Blueprint, request
from flask.views import MethodView
from extensions import db
from models.article import Article
from utils.response import success, fail
class ArticleListView(MethodView):
"""集合视角:/api/article —— GET 取列表、POST 新增"""
def get(self):
page = request.args.get("page", 1, type=int)
page_size = request.args.get("page_size", 10, type=int)
category_id = request.args.get("category_id", type=int)
dynasty = request.args.get("dynasty")
keyword = request.args.get("keyword")
order_by = request.args.get("order_by", "latest")
query = Article.query.filter(Article.status == 1)
if category_id:
query = query.filter(Article.category_id == category_id)
if dynasty:
query = query.filter(Article.dynasty == dynasty)
if keyword:
like = f"%{keyword}%"
query = query.filter(db.or_(Article.title.like(like),
Article.summary.like(like)))
if order_by == "hot":
query = query.order_by(Article.views.desc(), Article.id.desc())
elif order_by == "likes":
query = query.order_by(Article.likes.desc(), Article.id.desc())
else:
query = query.order_by(Article.created_at.desc(), Article.id.desc())
page_obj = db.paginate(query, page=page, per_page=page_size, error_out=False)
return success({
"list": [a.to_list_dict() for a in page_obj.items],
"total": page_obj.total,
"page": page,
"page_size": page_size,
})
def post(self):
data = request.get_json(silent=True) or {}
title = (data.get("title") or "").strip()
if not title:
return fail("缺少参数:title")
article = Article(title=title,
author=data.get("author"),
dynasty=data.get("dynasty"),
category_id=data.get("category_id"),
summary=data.get("summary"),
content=data.get("content"),
status=1)
db.session.add(article)
db.session.commit()
return success({"id": article.id}, "新增成功")
class ArticleDetailView(MethodView):
"""成员视角:/api/article/<aid> —— GET 详情、PUT 修改、DELETE 删除"""
def get(self, aid):
article = db.session.get(Article, aid)
if article is None or article.status != 1:
return fail("作品不存在", code=404)
article.views = (article.views or 0) + 1
db.session.commit()
return success(article.to_detail_dict())
def put(self, aid):
article = db.session.get(Article, aid)
if article is None:
return fail("作品不存在", code=404)
data = request.get_json(silent=True) or {}
for field in ("title", "author", "dynasty",
"category_id", "summary", "content"):
if field in data:
setattr(article, field, data[field])
db.session.commit()
return success(None, "修改成功")
def delete(self, aid):
article = db.session.get(Article, aid)
if article is None:
return fail("作品不存在", code=404)
article.status = 0 # 软删除:数据留着,前端查不到了
db.session.commit()
return success(None, "删除成功")
article_bp = Blueprint("article", __name__, url_prefix="/api/article")
# 集合:GET /api/article/list 查列表,POST /api/article 新增
article_bp.add_url_rule("/list",
view_func=ArticleListView.as_view("article_list"),
methods=["GET"])
article_bp.add_url_rule("",
view_func=ArticleListView.as_view("article_create"),
methods=["POST"])
# 成员:GET 详情、PUT 修改、DELETE 删除,共用同一个实例逻辑
article_bp.add_url_rule("/detail/<int:aid>",
view_func=ArticleDetailView.as_view("article_detail"),
methods=["GET"])
article_bp.add_url_rule("/<int:aid>",
view_func=ArticleDetailView.as_view("article_item"),
methods=["PUT", "DELETE"])
from flask import Blueprint, g, request
from flask.views import MethodView
from extensions import db
from models.article import Article
from models.comment import Comment
from utils.auth import login_required
from utils.response import fail, success
comment_bp = Blueprint("comment", __name__, url_prefix="/api/comment")
class CommentActionView(MethodView):
"""书评的写操作:整类都要登录,写在 decorators 里一次搞定"""
decorators = [login_required]
def post(self):
data = request.get_json(silent=True) or {}
article_id = data.get("article_id")
content = (data.get("content") or "").strip()
if not article_id:
return fail("缺少参数:article_id")
if not content:
return fail("书评内容不能为空")
if db.session.get(Article, article_id) is None:
return fail("作品不存在", code=404)
comment = Comment(user_id=g.current_user.id,
article_id=article_id,
parent_id=data.get("parent_id"),
content=content)
db.session.add(comment)
db.session.commit()
return success({"comment_id": comment.id}, "发表成功")
def delete(self, cid):
comment = db.session.get(Comment, cid)
if comment is None or comment.status != 1:
return fail("书评不存在", code=404)
if comment.user_id != g.current_user.id:
return fail("无权删除他人书评", code=403)
comment.status = 0
db.session.commit()
return success(None, "删除成功")
comment_bp.add_url_rule("/add",
view_func=CommentActionView.as_view("comment_add"),
methods=["POST"])
comment_bp.add_url_rule("/<int:cid>",
view_func=CommentActionView.as_view("comment_delete"),
methods=["DELETE"])
# ---------- 第 1 题 ----------
from flask.views import MethodView
from models.article import Banner # 轮播图模型
from utils.response import success
class BannerListView(MethodView):
def get(self):
rows = (Banner.query
.filter_by(status=1)
.order_by(Banner.sort.asc())
.all())
return success([b.to_dict() for b in rows])
app.add_url_rule("/api/banner/list",
view_func=BannerListView.as_view("banner_list"))
# Methods 会显示 GET, HEAD, OPTIONS。
# 因为类里只定义了 get(),MethodView 自动推导出的方法列表就只有 GET;
# HEAD 与 OPTIONS 是 Werkzeug 给每条规则自动加上的。
# ---------- 第 2 题 ----------
from flask import Blueprint, g, request
from flask.views import MethodView
from extensions import db
from models.article import Article
from models.comment import Favorite
from utils.auth import login_required
from utils.response import fail, success
favorite_bp = Blueprint("favorite", __name__, url_prefix="/api/favorite")
class FavoriteView(MethodView):
decorators = [login_required]
def get(self):
page = request.args.get("page", 1, type=int)
page_size = request.args.get("page_size", 10, type=int)
query = Favorite.query.filter_by(user_id=g.current_user.id) \
.order_by(Favorite.created_at.desc())
page_obj = db.paginate(query, page=page, per_page=page_size, error_out=False)
return success({
"list": [f.to_dict() for f in page_obj.items],
"total": page_obj.total,
"page": page,
"page_size": page_size,
})
def post(self):
data = request.get_json(silent=True) or {}
article_id = data.get("article_id")
if not article_id:
return fail("缺少参数:article_id")
if db.session.get(Article, article_id) is None:
return fail("作品不存在", code=404)
row = Favorite.query.filter_by(user_id=g.current_user.id,
article_id=article_id).first()
if row: # 已收藏 → 取消
db.session.delete(row)
favorited = False
else: # 未收藏 → 收藏
db.session.add(Favorite(user_id=g.current_user.id, article_id=article_id))
favorited = True
db.session.commit()
return success({"favorited": favorited}, "操作成功")
favorite_bp.add_url_rule("/list",
view_func=FavoriteView.as_view("favorite_list"),
methods=["GET"])
favorite_bp.add_url_rule("/toggle",
view_func=FavoriteView.as_view("favorite_toggle"),
methods=["POST"])
# ---------- 第 3 题(红影赏析,换的是模型,骨架完全一致)----------
class MovieListView(MethodView):
def get(self):
movies = Movie.query.order_by(Movie.rating.desc()).all()
return success([m.to_dict() for m in movies])
def post(self):
data = request.get_json(silent=True) or {}
if not (data.get("title") or "").strip():
return fail("缺少参数:title")
movie = Movie(title=data["title"],
genre=data.get("genre"),
rating=data.get("rating"))
db.session.add(movie)
db.session.commit()
return success({"id": movie.id}, "新增成功")
class MovieDetailView(MethodView):
def get(self, mid):
movie = db.session.get(Movie, mid)
if movie is None:
return fail("电影不存在", code=404)
return success(movie.to_dict())
def delete(self, mid):
movie = db.session.get(Movie, mid)
if movie is None:
return fail("电影不存在", code=404)
db.session.delete(movie)
db.session.commit()
return success(None, "删除成功")
第 3 题的关键:换掉的是「模型名和字段名」,不变的是「集合类管 list/create、成员类管 detail/update/delete」这套骨架。第 2 题里的 toggle 逻辑(先查存在与否 → 存在就删、不存在就加)是下一章的主角,先记住这个套路。