（九）为一次请求安排存储与事务

1. 每条书籍关联一位用户

第五章已经用连接和事务保存书籍。现在添加用户归属：书名在同一用户的书架内唯一，两个用户可以分别保存同名书。

# 项目文件：web_project/src/webbook/database.py
# 请求各自取得连接；事务退出后提交或回滚，finally再关闭连接。
import csv
import json
import sqlite3
from contextlib import contextmanager
from pathlib import Path

from werkzeug.security import generate_password_hash

from .validation import validate_book

SCHEMA = """
CREATE TABLE IF NOT EXISTS users(
 id INTEGER PRIMARY KEY, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS sessions(
 sid TEXT PRIMARY KEY, csrf TEXT NOT NULL, user_id INTEGER REFERENCES users(id), expires INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS books(
 id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id),
 title TEXT NOT NULL, finished INTEGER NOT NULL CHECK(finished IN (0,1)), tags TEXT NOT NULL,
 UNIQUE(user_id,title)
);
"""


@contextmanager
def connect(path):
    connection = sqlite3.connect(path, timeout=1)
    connection.row_factory = sqlite3.Row
    connection.execute("PRAGMA foreign_keys=ON")
    try:
        with connection:
            yield connection
    finally:
        connection.close()

books 的 user_id 指向 users，UNIQUE(user_id,title) 由数据库判断同账号同名冲突；sessions 中的 user_id 可以为空，用来表示尚未登录但已经打开表单的会话。

connect 每次取得独立连接，with connection 处理事务提交或回滚，finally 再关闭连接。这两个动作分开：SQLite 连接的事务上下文结束，不等于连接对象已经关闭。

用户记录关联各自书籍，会话再关联用户；一次请求取得连接进入事务，结束后提交或回滚并关闭

2. 从采集 CSV 初始化已有记录

# 续写同一文件：web_project/src/webbook/database.py
# 初始化时先查账户是否存在，再导入书籍；重复执行不会反复创建同一账户和记录。
def initialize(path: Path, source=None):
    path.parent.mkdir(parents=True, exist_ok=True)
    with connect(path) as connection:
        connection.executescript(SCHEMA)
        for username, password in (("alice", "learn-alice"), ("bob", "learn-bob")):
            if (
                connection.execute(
                    "SELECT id FROM users WHERE username=?", (username,)
                ).fetchone()
                is None
            ):
                connection.execute(
                    "INSERT INTO users(username,password_hash) VALUES (?,?)",
                    (username, generate_password_hash(password)),
                )
        if source is not None:
            owner = connection.execute(
                "SELECT id FROM users WHERE username='alice'"
            ).fetchone()[0]
            with source.open(encoding="utf-8-sig", newline="") as stream:
                for row in csv.DictReader(stream):
                    if row["finished"] not in {"0", "1"}:
                        raise ValueError("CSV中的finished应为0或1")
                    book = validate_book(
                        {
                            "title": row["title"],
                            "finished": row["finished"] == "1",
                            "tags": json.loads(row["tags"]),
                        }
                    )
                    connection.execute(
                        "INSERT OR IGNORE INTO books(user_id,title,finished,tags) VALUES (?,?,?,?)",
                        (
                            owner,
                            book.title,
                            int(book.finished),
                            json.dumps(book.tags, ensure_ascii=False),
                        ),
                    )

初始化先建立账号，再将前面 CSV 的 finished 文本转换为布尔值、tags 文本转换为列表，交给同一个 validate_book。已有同名书不重复新增，再次初始化也不覆盖已有密码摘要。

SQL 的问号继续用于传入数据。SQL 结构由程序确定，书名、用户名等值交给参数绑定，不把用户文本拼进语句中。

3. 业务方法保留同一套含义

# 项目文件：web_project/src/webbook/service.py
# 用户编号来自会话，不从表单接受；占位符传值，唯一约束最终判断重复书名。
import json
import sqlite3

from .http import Problem


def as_book(row):
    return {
        "id": row["id"],
        "title": row["title"],
        "finished": bool(row["finished"]),
        "tags": json.loads(row["tags"]),
    }


def list_books(connection, owner, finished=None):
    rows = connection.execute(
        """SELECT * FROM books WHERE user_id=?
                                 AND (? IS NULL OR finished=?) ORDER BY id""",
        (owner, finished, finished),
    ).fetchall()
    return [as_book(row) for row in rows]


def find_book(connection, book_id):
    row = connection.execute("SELECT * FROM books WHERE id=?", (book_id,)).fetchone()
    if row is None:
        raise Problem(404, "没有找到这本书")
    return row


def owned_book(connection, book_id, owner):
    row = find_book(connection, book_id)
    if row["user_id"] != owner:
        raise Problem(403, "这本书不属于当前账号")
    return row

list_books 在查询时就按 owner 筛选，再按需筛选已读状态。find_book 处理没有找到的记录；owned_book 再判断这条记录属于谁。book_id 从路径取得的数字文本作为参数交给查询，读取结果里的 id 才是数据库中的整数编号。

# 续写同一文件：web_project/src/webbook/service.py
# 创建时把同名冲突转成 409；修改已读状态前先核对这本书是否属于当前账户。
def create_book(connection, owner, book):
    try:
        cursor = connection.execute(
            "INSERT INTO books(user_id,title,finished,tags) VALUES (?,?,?,?)",
            (
                owner,
                book.title,
                int(book.finished),
                json.dumps(book.tags, ensure_ascii=False),
            ),
        )
    except sqlite3.IntegrityError as error:
        if error.sqlite_errorname == "SQLITE_CONSTRAINT_UNIQUE":
            raise Problem(409, "这位用户已经保存了同名书籍") from error
        raise
    return as_book(find_book(connection, cursor.lastrowid))


def finish_book(connection, owner, book_id):
    owned_book(connection, book_id, owner)
    connection.execute("UPDATE books SET finished=1 WHERE id=?", (book_id,))
    return as_book(find_book(connection, book_id))

新增遇到同用户同名的唯一约束时转为 409；其他数据库异常继续向上交给请求入口处理。标记已读先确认归属，再更新，两个操作处于同一个请求事务内。