（十一）把读取与写入保留在业务函数
1. 查询记录和核对归属
先把数据库行变成普通业务数据：
# 续写同一文件：shelf_project/src/shelfweb/db.py
def public_book(row):
# API 和模板使用普通字典，不把数据库连接留给模板迭代。
return {
"id": row["id"],
"title": row["title"],
"finished": bool(row["finished"]),
"tags": json.loads(row["tags"]),
}# 项目文件：shelf_project/src/shelfweb/service.py
import json
import sqlite3
from flask import abort
from .db import get_db, public_book
def list_books(owner_id, q="", finished="all"):
clauses, parameters = ["owner_id=?"], [owner_id]
if q:
# SQL 参数绑定和 LIKE 通配符转义分别处理不同的解释过程。
escaped = q.replace("!", "!!").replace("%", "!%").replace("_", "!_")
clauses.append("title LIKE ? ESCAPE '!'")
parameters.append(f"%{escaped}%")
if finished != "all":
clauses.append("finished=?")
parameters.append(int(finished))
query = "SELECT * FROM books WHERE " + " AND ".join(clauses) + " ORDER BY id"
return [public_book(row) for row in get_db().execute(query, parameters).fetchall()]
def owned_book(book_id, owner_id):
if book_id > 2**63 - 1:
abort(404, description="没有找到这本书")
row = get_db().execute("SELECT * FROM books WHERE id=?", (book_id,)).fetchone()
if row is None:
abort(404, description="没有找到这本书")
if row["owner_id"] != owner_id:
abort(403, description="这本书属于其他用户")
return public_book(row)列表查询先加 owner_id 条件，再按书名和已读状态筛选，最后按编号排序。clauses 中的 SQL 片段由程序提供，用户文本放进参数列表；LIKE 里的 %、_ 另外用 ! 转义，使输入百分号时查找实际包含百分号的书名。
owned_book 对找不到的记录返回 404，对其他用户的书返回 403。路径编号很大时，也统一作为未找到处理，再按可保存的数据库编号查询。public_book 稍后将行转成普通字典，恢复标签 JSON 和布尔状态，供页面和 JSON 接口复用。
2. 在成功返回之前完成事务
# 续写同一文件：shelf_project/src/shelfweb/service.py
def save_book(owner_id, draft, book_id=None):
connection = get_db()
try:
# 退出 with 时提交；SQL 或后续操作出错则回滚。
with connection:
fields = (
draft.title,
int(draft.finished),
json.dumps(draft.tags, ensure_ascii=False),
)
if book_id is None:
cursor = connection.execute(
"INSERT INTO books(title,finished,tags,owner_id) VALUES (?,?,?,?)",
(*fields, owner_id),
)
book_id = cursor.lastrowid
else:
# 地址中的编号需要核对归属，实际 UPDATE 也携带用户条件。
owned_book(book_id, owner_id)
connection.execute(
"UPDATE books SET title=?,finished=?,tags=? WHERE id=? AND owner_id=?",
(*fields, book_id, owner_id),
)
except sqlite3.IntegrityError as exc:
if exc.sqlite_errorname == "SQLITE_CONSTRAINT_UNIQUE":
# 事务已经回滚，视图用字段错误重新显示用户这次的尝试值。
return None, {"title": "书架里已经有同名的书"}
raise
return book_id, {}
def delete_book(book_id, owner_id):
# 未找到或归属不符时，在 DELETE 执行之前结束。
owned_book(book_id, owner_id)
connection = get_db()
with connection:
connection.execute(
"DELETE FROM books WHERE id=? AND owner_id=?", (book_id, owner_id)
)新增、修改和删除进入 with connection:。正常结束提交，异常离开回滚；这个 with 管理事务，实际关闭连接由 teardown_appcontext 完成。
fields 统一准备书名、整数状态和标签 JSON，(*fields, owner_id) 使用前面已经学过的拆包构造参数。新增后从 cursor.lastrowid 取得编号；修改和删除都核对归属，并在实际 SQL 中再次带上 owner_id 条件。
同名冲突在离开事务之后返回 errors，页面可以显示“已经有同名的书”。其他数据库异常继续抛出，交给后面的统一处理器。
