SQLite、事务与数据访问

SQLite 让应用获得事务、约束和查询能力，同时不需要独立数据库服务。它适合本地工具、原型和中小规模单机应用，但仍然需要认真设计表结构、参数化查询和事务边界。

一、学习目标

使用 sqlite3 建表并启用约束
始终使用参数化查询
理解事务提交与回滚
把数据库行转换为领域对象
用仓储隔离 SQL 与业务逻辑

二、建立连接

import sqlite3
from pathlib import Path

connection = sqlite3.connect(Path("tasks.db"))
connection.row_factory = sqlite3.Row
connection.execute("PRAGMA foreign_keys = ON")

row_factory 让查询结果可按字段名读取。外键约束需要对每个连接显式启用。

连接也可用作事务上下文：

with connection:
    connection.execute(...)

块内发生异常时回滚，否则提交。离开块不会自动关闭连接，仍需调用 close() 或由仓储上下文管理器负责。

三、创建表

connection.execute(
    """
    CREATE TABLE IF NOT EXISTS tasks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL CHECK (length(trim(title)) > 0),
        status TEXT NOT NULL CHECK (status IN ('todo', 'done')),
        created_at TEXT NOT NULL,
        updated_at TEXT NOT NULL
    )
    """
)
connection.commit()

数据库约束是最后一道防线：

NOT NULL 防止缺失值；
CHECK 防止空标题和未知状态；
主键保证唯一身份。

应用校验提供友好错误，数据库约束防止其他写入路径破坏数据。

四、参数化查询

cursor = connection.execute(
    """
    INSERT INTO tasks (title, status, created_at, updated_at)
    VALUES (?, ?, ?, ?)
    """,
    (task.title, task.status.value, task.created_at.isoformat(), task.updated_at.isoformat()),
)

参数必须单独传递，不能用 f-string 拼进 SQL：

# 错误：存在 SQL 注入和引号问题
connection.execute(f"SELECT * FROM tasks WHERE title = '{title}'")

表名、列名等结构不能使用普通值参数；如果确实动态选择，必须从程序内白名单映射。

五、读取与映射

from datetime import datetime


def to_task(row: sqlite3.Row) -> Task:
    return Task(
        id=int(row["id"]),
        title=str(row["title"]),
        status=TaskStatus(str(row["status"])),
        created_at=datetime.fromisoformat(str(row["created_at"])),
        updated_at=datetime.fromisoformat(str(row["updated_at"])),
    )

查询：

rows = connection.execute(
    """
    SELECT id, title, status, created_at, updated_at
    FROM tasks
    WHERE status = ?
    ORDER BY id
    """,
    (TaskStatus.TODO.value,),
)
tasks = [to_task(row) for row in rows]

显式列出字段比 SELECT * 更稳定，也更容易审查。

六、仓储边界

class TaskRepository:
    def __init__(self, database: Path) -> None:
        self._connection = sqlite3.connect(database)
        self._connection.row_factory = sqlite3.Row
        self._create_schema()

    def add(self, task: Task) -> Task:
        with self._connection:
            cursor = self._connection.execute(
                """
                INSERT INTO tasks (title, status, created_at, updated_at)
                VALUES (?, ?, ?, ?)
                """,
                (
                    task.title,
                    task.status.value,
                    task.created_at.isoformat(),
                    task.updated_at.isoformat(),
                ),
            )
        if cursor.lastrowid is None:
            raise RuntimeError("数据库未返回任务 ID")
        return self.get(cursor.lastrowid)

服务层只调用 add、list、set_status 等领域操作，不拼 SQL。

七、事务边界

完成任务并写审计记录必须一起成功：

with connection:
    updated = connection.execute(
        "UPDATE tasks SET status = 'done', updated_at = ? WHERE id = ?",
        (now.isoformat(), task_id),
    )
    if updated.rowcount != 1:
        raise KeyError(f"任务 {task_id} 不存在")
    connection.execute(
        "INSERT INTO task_events (task_id, action, created_at) VALUES (?, ?, ?)",
        (task_id, "completed", now.isoformat()),
    )

如果第二条语句失败，第一条也回滚。

事务不应包住等待用户输入或网络请求，否则会长期持有锁。

八、索引与查询计划

当数据增长后，可为常用筛选与排序建立索引：

CREATE INDEX IF NOT EXISTS tasks_status_id_idx
ON tasks (status, id);

检查查询计划：

EXPLAIN QUERY PLAN
SELECT id, title FROM tasks WHERE status = 'todo' ORDER BY id;

不要为每列都建索引。索引加速部分读取，但增加写入和存储成本，应由真实查询模式驱动。

九、并发与适用边界

SQLite 允许多个读取者，但写入仍需要协调。短事务、本地磁盘和合理超时很重要。以下情况应考虑服务型数据库：

多台应用服务器并发写入；
持续高写入吞吐；
复杂权限、复制或高可用需求；
需要专门运维与监控能力。

不要把 SQLite 数据库放在不保证锁语义的共享网络文件系统上。

十、常见错误

通过字符串拼接构建 SQL。
忘记提交事务或依赖不清晰的隐式行为。
捕获数据库异常后继续使用已失败的事务。
在循环中逐条查询，形成 N+1。
业务层到处直接操作连接。

十一、练习与自测

实现按状态筛选与完成任务。
为状态与 ID 建联合索引并检查查询计划。
模拟审计写入失败，确认任务状态回滚。

自测：

参数化查询能防止哪类问题？
with connection 是否会关闭连接？
数据库约束与应用校验为何需要同时存在？

十二、官方资料

上一篇：命令行、配置与日志｜下一篇：HTTP、JSON 与 API 客户端