SQLite的零配置与单文件存储特性让它在原型、桌面端和边缘设备中很常见。构建RESTful API时，如果直接把它当成一个临时的数据文件随意读写，会出现锁冲突、连接泄漏和错误响应不一致等问题。本文围绕一个任务管理服务展开，展示如何从表结构、路由、事务和错误处理几个层面把SQLite嵌入可靠的API。

表结构设计与资源建模

在设计REST API之前，先理清资源边界。任务管理服务包含两种资源：任务和任务操作日志。任务资源对应 /tasks 路径，日志资源只作为内部审计数据，不直接通过REST暴露。SQLite中建表时需要利用其动态类型系统的灵活性，但不要完全放弃约束。title字段用TEXT加CHECK约束限制长度，status字段用CHECK枚举值保证数据一致性，避免应用层疏漏写入脏数据。任务表主键使用INTEGER PRIMARY KEY AUTOINCREMENT，这是SQLite中推荐的自增键写法，实际上INTEGER PRIMARY KEY本身就能复用rowid，AUTOINCREMENT会增加额外的sqlite_sequence开销，除非需要防止ID复用，否则可以省略AUTOINCREMENT。

时间字段使用datetime('now')，但updated_at不能自动更新，需要在UPDATE语句中显式赋值，或者创建触发器。外键默认关闭，需要每次连接执行PRAGMA foreign_keys=ON，否则task_log的REFERENCES不会生效。索引方面为status创建普通索引，如果经常按created_at排序，也可以为created_at建立索引。资源建模上，让数据库表的行对应API的JSON对象，表字段名与JSON属性名保持一致，序列化时直接用sqlite3.Row和dict转换，减少手动映射层。这个做法适合中小型项目，如果API需要向外部暴露不同的字段形状，可以再增加DTO层，而不是在SQL里做大量别名。初始化脚本如下：

import sqlite3
def init_db(path):
    conn = sqlite3.connect(path)
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA foreign_keys=ON;")
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS tasks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL CHECK(length(title) BETWEEN 1 AND 200),
            description TEXT DEFAULT '',
            status TEXT NOT NULL DEFAULT 'pending'
                CHECK(status IN ('pending','done','archived')),
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            updated_at TEXT NOT NULL DEFAULT (datetime('now'))
        );
        CREATE TABLE IF NOT EXISTS task_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            task_id INTEGER NOT NULL REFERENCES tasks(id) ON DELETE CASCADE,
            action TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT (datetime('now'))
        );
        CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
    """)
    conn.commit()
    conn.close()

这段代码同时开启了WAL模式和外键约束，并创建了两张表。WAL模式对整个数据库文件生效，应在每个连接上设置或者只在初始化时设置一次即可，因为PRAGMA journal_mode会持久化到数据库文件中。tasks表的CHECK约束会阻止超长标题和非法状态写入，即使应用层忘记校验，数据库层也能兜底。task_log表使用外键关联tasks，删除任务时如果启用外键，日志会级联删除；如果希望保留历史日志，可以改为SET NULL或干脆不建外键，但需要应用层维护一致性。

路由层与参数化查询

Flask或类似微框架适合与SQLite搭配。每个请求开始时创建连接，请求结束时关闭，避免线程间共享连接导致游标状态串扰。get_db函数使用flask.g保存连接对象，teardown_appcontext中调用close。连接上设置row_factory = sqlite3.Row，让查询结果能按列名访问并方便转成字典。添加PRAGMA busy_timeout=5000，让数据库遇到写锁时等待5秒而不是立刻抛出database is locked，这对低频API足够。

路由层坚持使用参数化查询，不要用字符串拼接用户输入。例如错误写法：db.execute(f"SELECT * FROM tasks WHERE title = '{title}'")，如果title包含单引号就会破坏SQL语法，甚至被注入。正确写法是使用问号占位符和参数元组：db.execute("SELECT * FROM tasks WHERE title = ?", (title,))。下面代码展示任务列表和创建接口：

from flask import Flask, request, jsonify, g
import sqlite3, os
app = Flask(__name__)
DATABASE = os.environ.get("DB_PATH", "tasks.db")
def get_db():
    if 'db' not in g:
        conn = sqlite3.connect(DATABASE)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA foreign_keys=ON;")
        conn.execute("PRAGMA busy_timeout=5000;")
        g.db = conn
    return g.db
@app.teardown_appcontext
def close_db(exc):
    conn = g.pop('db', None)
    if conn is not None:
        conn.close()
@app.route('/tasks', methods=['GET'])
def list_tasks():
    db = get_db()
    status = request.args.get('status')
    if status:
        rows = db.execute("SELECT * FROM tasks WHERE status = ? ORDER BY id DESC", (status,)).fetchall()
    else:
        rows = db.execute("SELECT * FROM tasks ORDER BY id DESC").fetchall()
    return jsonify([dict(row) for row in rows])
@app.route('/tasks', methods=['POST'])
def create_task():
    data = request.get_json(silent=True) or {}
    title = data.get('title', '').strip()
    if not title or len(title) > 200:
        return jsonify({"error": "title is required and must be under 200 characters"}), 400
    db = get_db()
    cur = db.execute(
        "INSERT INTO tasks (title, description, status) VALUES (?, ?, ?)",
        (title, data.get('description', ''), data.get('status', 'pending'))
    )
    db.commit()
    row = db.execute("SELECT * FROM tasks WHERE id = ?", (cur.lastrowid,)).fetchone()
    return jsonify(dict(row)), 201

GET /tasks可以根据status查询参数过滤，使用WHERE子句后接参数元组，而不是拼接字符串。创建接口需要显式校验title长度，数据库CHECK约束是最后防线，但应用层应返回明确的400错误给客户端，而不是让数据库异常直接暴露。获取JSON时使用silent=True防止客户端未发送Content-Type时抛异常。插入后使用cur.lastrowid获取新ID，再查询一次返回完整对象，虽然多一次查询，但能确保返回结果包含默认值和时间戳。如果在意性能，可以在插入前生成时间戳并指定所有字段，但本例关注清晰度。分页可以在后续通过LIMIT和OFFSET添加，这两个子句不能参数化，需要用整数解析后插入到SQL中并做边界限制，例如page_size不超过100。

并发控制与事务边界

SQLite的锁粒度是整个数据库，写操作会独占写锁，读操作在默认回滚日志模式下允许并发读，但写事务提交时可能阻塞其他读写。为了改善读写并发，连接时执行PRAGMA journal_mode=WAL，将日志写入独立的WAL文件，读事务不再阻塞写事务，写事务也不会阻塞读，适合REST API中读取列表和写入单条记录交替的场景。必须设置busy_timeout，否则多线程同时写入时，第二个连接立即失败。多进程部署时，WAL模式仍然有效，但同一时刻只有一个进程可以写入，写吞吐量受限于单文件，如果需要高并发写入，应尽早迁移到PostgreSQL等服务器数据库。

事务边界方面，只读操作不必显式开启事务，SQLite会为单条SELECT自动使用读事务。对于多步写操作，如归档任务并记录日志，两个写入操作必须放在同一个事务里保证原子性。推荐使用with db:语法自动处理提交和回滚，而不是手动try/except。下面的代码展示一个归档接口，使用with表达式包裹两个操作，如果第二个SQL失败，第一个也会回滚。

@app.route('/tasks/<int:task_id>/archive', methods=['POST'])
def archive_task(task_id):
    db = get_db()
    try:
        with db:
            row = db.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()
            if row is None:
                return jsonify({"error": "task not found"}), 404
            db.execute("UPDATE tasks SET status = 'archived', updated_at = datetime('now') WHERE id = ?", (task_id,))
            db.execute("INSERT INTO task_log (task_id, action) VALUES (?, 'archived')", (task_id,))
    except sqlite3.Error as exc:
        return jsonify({"error": str(exc)}), 500
    return jsonify({"message": "task archived"})

with db:对应db对象的上下文管理器，会在块结束时自动commit或rollback。如果使用手动事务，务必在except中调用db.rollback()，否则连接会持续处于事务中，后续操作可能出现逻辑错误。另一个常见错误是在应用启动时只打开一个全局连接，然后多个线程共享，SQLite连接默认不保证跨线程安全，会出现cursor状态错乱。正确做法是每请求创建短连接，配合WAL和busy_timeout。对于只读副本和备份，可以使用sqlite3.Connection.backup方法在线备份，不必停止API。

错误处理与状态码语义

REST API的错误响应结构应保持一致。本案例采用{"error": "描述信息"}的简单格式，客户端根据HTTP状态码判断错误类别。SQLite的IntegrityError主要来自唯一约束和CHECK约束，例如重复插入相同唯一字段或status值非法。在应用层应捕获这类异常并转换为409 Conflict，而不是返回500。资源不存在时返回404，创建成功返回201并在响应体返回新建对象，删除成功返回204且响应体为空。

统一错误处理可以通过Flask的errorhandler注册，但需要注意errorhandler捕获的异常必须发生在请求处理过程中。视图函数中如果已经用with db:提交事务，IntegrityError会在提交时触发，此时仍能被errorhandler捕获。若在事务块内部显式捕获，则需手动返回。下面的代码展示了一个简单的全局异常处理器：

import sqlite3
from flask import jsonify
@app.errorhandler(sqlite3.IntegrityError)
def handle_integrity_error(exc):
    return jsonify({"error": "data conflict"}), 409
@app.errorhandler(404)
def handle_not_found(exc):
    return jsonify({"error": "resource not found"}), 404

日志记录也不能省，可以使用logging.exception将原始异常写入服务端日志，但不要把堆栈返回给客户端，否则可能泄露表结构信息。对于请求体解析失败，比如客户端发送了非JSON格式的Content-Type，可以返回400和明确的错误信息。状态码的使用要稳定，不要用200包裹错误信息，那样会破坏HTTP语义，客户端难以判断是否真正成功。将错误响应格式固定下来，客户端只需检查状态码和error字段，不用解析不同的异构结构。

测试与部署建议

SQLite在测试阶段非常友好，因为不需要独立数据库服务。pytest的tmp_path可以创建临时文件路径，每个测试用例使用独立的数据库文件，避免测试间数据污染。编写fixture时，创建应用实例，设置TESTING和DATABASE路径，调用init_db，然后用app.test_client()发起请求。测试重点应覆盖资源生命周期和错误路径，例如创建任务后按ID读取、更新、删除，以及不存在的ID返回404。

import os, tempfile, pytest
from app import app
from init_db import init_db
@pytest.fixture
def client(tmp_path):
    app.config['TESTING'] = True
    app.config['DATABASE'] = str(tmp_path / 'test.db')
    with app.app_context():
        init_db(app.config['DATABASE'])
    with app.test_client() as client:
        yield client
def test_create_and_get_task(client):
    resp = client.post('/tasks', json={'title': '学习SQLite'})
    assert resp.status_code == 201
    task_id = resp.get_json()['id']
    resp = client.get(f'/tasks/{task_id}')
    assert resp.status_code == 200
    assert resp.get_json()['title'] == '学习SQLite'
def test_missing_task_returns_404(client):
    resp = client.get('/tasks/9999')
    assert resp.status_code == 404

测试中不要mock sqlite3层，直接使用真实数据库能更早发现SQL语法和约束问题，运行速度也足够快。部署时需要注意数据库文件所在目录对进程有写权限，因为SQLite会创建WAL和SHM文件。如果使用systemd或容器，确保工作目录正确。备份可以使用sqlite3的backup API或直接复制WAL checkpoint之后的文件，但不要在API运行中直接复制主文件，可能得到不一致快照。定期执行PRAGMA wal_checkpoint(TRUNCATE)控制WAL文件大小，避免磁盘占用增长。整体上，SQLite适合单机、中小规模、读多写少的API场景，将其视作认真的数据层而不是临时存储，配合WAL、参数化、事务和统一错误处理，可以构建出可靠且易维护的服务。