Flask 笔记十：把查询逻辑抽到 service，让 views 变薄

上一篇我们做了登录、Session 和@login_required。路由能保护了，但views.py往往还会越来越长：读参数、拼 SQL、分页、再render_template全挤在一个函数里。

这一篇做一件事：把「怎么查数据」从视图里挪出去，视图只负责「读请求 → 调函数 → 选模板」。

例子仍是通用的Note备忘录，不涉及任何真实业务。

1. 学完后你能做什么

• 分清 视图该写什么、service 该写什么
• 新建note_service.py，把列表查询抽成函数
• 登录后 「只看自己的备忘录」 也放在 service 里
• 同一个查询函数，列表页和导出 可以共用
• 知道什么时候 不必再抽一层

2. 视图变胖，通常长什么样

第五篇你可能已经写过类似代码：

@home.route("/notes/")

@login_required

def note_list():

q = (request.args.get("q") or "").strip()

page = request.args.get("page", 1, type=int)

user_id = session.get("user_id")

query = Note.query.filter_by(user_id=user_id).order_by(Note.addtime.desc())

if q:

like = f"%{q}%"

query = query.filter(

or_(Note.title.like(like), Note.content.like(like))

)

page_data = query.paginate(page=page, per_page=10)

return render_template(

"home/note_list.html",

page_data=page_data,

delete_form=DeleteForm(),

q=q,

)

能跑。但再加「日期筛选」「置顶优先」「admin 后台也要同逻辑」时，这段查询会 复制粘贴好几份，改一处漏一处。

问题不在 SQLAlchemy，而在 职责混在一起：

3. 先建app/note_service.py

新建文件，专门放和Note有关的查询：

from sqlalchemy import or_

from app.models import Note

def list_notes_for_user(

user_id: int,

*,

q: str = "",

date_from: str = "",

date_to: str = "",

page: int = 1,

per_page: int = 10,

):

"""某用户的备忘录列表（支持搜索、日期、分页）。"""

query = (

Note.query

.filter_by(user_id=user_id)

.order_by(Note.addtime.desc())

)

q = (q or "").strip()

if q:

like = f"%{q}%"

query = query.filter(

or_(Note.title.like(like), Note.content.like(like))

)

date_from = (date_from or "").strip()

date_to = (date_to or "").strip()

if date_from:

query = query.filter(Note.addtime >= date_from)

if date_to:

query = query.filter(Note.addtime <= date_to + " 23:59:59")

return query.paginate(page=page, per_page=per_page)

几个习惯：

• 关键字参数（*, q=...）：调用时一眼能看出传了什么
• 返回数据（这里是page_data），不render_template
• 函数名说清用途：list_notes_for_user，不是含糊的get_notes

4. 视图变薄

app/home/views.py：

from flask import request, session, render_template

from app.auth_utils import login_required

from app.forms import DeleteForm

from app.note_service import list_notes_for_user

@home.route("/notes/")

@login_required

def note_list():

q = (request.args.get("q") or "").strip()

date_from = (request.args.get("date_from") or "").strip()

date_to = (request.args.get("date_to") or "").strip()

page = request.args.get("page", 1, type=int)

page_data = list_notes_for_user(

session["user_id"],

q=q,

date_from=date_from,

date_to=date_to,

page=page,

)

return render_template(

"home/note_list.html",

page_data=page_data,

delete_form=DeleteForm(),

q=q,

date_from=date_from,

date_to=date_to,

)

对比之前：中间一大段 SQL 没了，读起来像目录——先读参数，再调 service，再渲染。

登录保护仍放在 视图 + 装饰器；service 假定「调用方已经知道 user_id」，不读session（后面会说为什么）。

5. 单条查询也抽出来

编辑、删除前都要「按 id 取一条，且必须是本人的」：

def get_note_for_user(note_id: int, user_id: int):

"""取一条备忘录；不存在或不属于该用户则返回 None。"""

return Note.query.filter_by(id=note_id, user_id=user_id).first()

编辑视图：

@home.route("/notes/edit/<int:note_id>/", methods=["GET", "POST"])

@login_required

def note_edit(note_id):

user_id = session["user_id"]

row = get_note_for_user(note_id, user_id)

if not row:

flash("记录不存在或无权访问", "err")

return redirect(url_for("home.note_list"))

form = NoteForm()

if request.method == "GET":

form.title.data = row.title

form.content.data = row.content

if form.validate_on_submit():

row.title = form.title.data.strip()

row.content = (form.content.data or "").strip()

db.session.commit()

flash("保存成功", "ok")

return redirect(url_for("home.note_list"))

return render_template("home/note_form.html", form=form, title="编辑备忘录")

比Note.query.get_or_404(note_id)更安全：别人的 id 不会误改，直接当「没有」处理。

6. service 为什么不读 session

新手常写：

def list_notes_for_user(...):

user_id = session.get("user_id") # 不推荐

短期省事，长期麻烦：

• 批处理脚本、定时任务没有 HTTP 请求，没有 session
• 单元测试要 mock session
• 同一个函数不好区分「查 A 用户」还是「查 B 用户」

更好做法：谁调用谁传user_id。视图从 session 取，脚本从参数取，service 只认数字 id。

鉴权（有没有登录）留在 装饰器 / 视图；数据归属（这条是不是你的）放在 service 或视图里显式传 user_id。

7. 一个查询，多处复用

以后若要 导出 CSV，不必复制 SQL：

from app.note_service import list_notes_for_user

def export_my_notes_csv(user_id):

# 不分页，取全量：per_page 设大，或另写 list_notes_for_user_all

page_data = list_notes_for_user(user_id, per_page=10000)

rows = page_data.items

# 写 CSV ...

列表页、导出、后台统计，共用同一套过滤规则，改搜索逻辑只改 service 一处。

8. 文件怎么摆（入门够用）

不必搞复杂目录，小项目常见：

app/

├── home/

│ └── views.py # 前台路由

├── admin/

│ └── views.py # 后台路由（下一篇可拆 Blueprint）

├── models.py

├── forms.py

├── auth_utils.py # login_required

├── note_service.py # Note 相关查询

└── user_service.py # User 相关（可选）

命名习惯：xxx_service.py或xxx_queries.py都行，团队统一即可。

一个文件对应一块业务，别把所有表的查询塞进一个service.py几千行。

9. 流程示意

GET /notes/?q=会议

│

▼

@login_required 确认已登录

│

▼

views.note_list 读 request.args、session["user_id"]

│

▼

list_notes_for_user(user_id, q="会议", ...)

│

▼

返回 paginate 结果（不碰模板）

│

▼

render_template("note_list.html", page_data=...)

GET /notes/edit/99/（别人的 id）

│

▼

get_note_for_user(99, my_user_id) → None

│

▼

flash + redirect（不暴露「有这条但你看不见」）

10. 新手常踩的 5 个坑

坑 1：service 里render_template

service 应 返回数据；渲染是视图的事。混在一起以后没法给 JSON API 复用。

坑 2：service 里flash/redirect

同上，属于 HTTP 层。service 返回None或抛自定义异常，由视图决定怎么提示用户。

坑 3：过度抽象

只有一条Note.query.get(id)，不必再包三层。 重复第二次时再抽。

坑 4：忘记在 service 里过滤user_id

登录只保证「你是谁」，不保证「你能动别人的数据」。写操作、按 id 查单条都要带 user_id。

坑 5：service 之间循环 import

note_service调user_service，user_service又调note_service会炸。
共用小逻辑可放models或utils.py；大模块之间尽量 单向依赖。

11. 和「大项目」的关系

真实项目里常见分层名字更多（Repository、DAO、Domain），但入门阶段记住一条就够：

视图处理 Web，service 处理「查什么、怎么查」。

你项目里若看到load_novel_chapter_view()、search_notes()这类函数，套路相同：视图短、查询集中、名字说清楚用途。

不必急着学 Application Factory 或复杂架构；先把重复的 SQL 从 views 挪出去，收益已经很大。

12. 小结

记住四件事：

1. views — 读参数、鉴权、redirect、render_template
2. service — 拼查询、分页、返回数据；不读 session
3. user_id显式传入 — 列表、单条查询都要管数据归属
4. 重复再抽 — 别为单行查询建十层抽象