ShowDocs
搜索文章Ctrl K

外观

← 返回SQLite列表
应用实践

SQLite：整数分账本完整 CLI ​

前置是参数、文件路径和事务。项目只包含标准库 Python 文件和依赖声明，无服务器或隐藏配置。适用 CPython 3.12+、其 sqlite3 所链接 SQLite >=3.37，核对 2026-10-17。本轮未执行任何示例，未创建数据库文件；以下结果是设计预期。

文件、依赖与运行顺序 ​

SQLite 账本 CLI 以 request 请求号防止重复记账，整数分 cents 必须为正。init 允许创建文件，其余操作要求原文件存在。下面命令由读者在 examples/sqlite 目录自行执行；Windows PowerShell 和 Linux shell 可使用同样参数（Linux 按需将 python 换成 python3）。requirements.txt 仅声明标准库能力，无 pip 产品依赖。

text
python ledger.py --db ledger.db init
python ledger.py --db ledger.db add r1 food 1250
python ledger.py --db ledger.db add r1 food 1250
python ledger.py --db ledger.db list
python ledger.py --db ledger.db plan
python ledger.py --db ledger.db backup ledger-copy.db

预期 init 输出 initialized=true；第一次 add 的 replayed=false，第二次 true；list 输出 [["food",1250,1]]。backup 输出目标绝对路径且不覆盖旧备份；plan 的具体文本依引擎和统计变化，不承诺固定结果。重复 request 改成 1300 应失败而不改变原记录。不要在教学文件以外运行写入命令。

唯一源码与参数绑定 ​

SQLite 参数绑定使用 ? 占位符，把 SQL 文本与独立参数分开。引号或 SQL 片段作为 category 进入时只是值，不会成为额外语句；占位符不能替代表名。本程序没有动态表名。BEGIN IMMEDIATE 和提交之间只有必要数据库操作，检查请求号与插入共享事务；异常时回滚后退出。

py
"""Reading-only teaching artifact in this delivery; no execution was performed."""
import argparse
import json
from pathlib import Path
import sqlite3
import sys


def connect(path, mode):
    uri = Path(path).resolve().as_uri() + "?mode=" + mode
    con = sqlite3.connect(uri, uri=True, isolation_level=None, timeout=2)
    con.execute("PRAGMA foreign_keys=ON")
    return con


def main():
    parser = argparse.ArgumentParser(description="Local integer-cent expense ledger")
    parser.add_argument("--db", default="ledger.db")
    sub = parser.add_subparsers(dest="command", required=True)
    sub.add_parser("init")
    add = sub.add_parser("add")
    add.add_argument("request")
    add.add_argument("category")
    add.add_argument("cents", type=int)
    sub.add_parser("list")
    sub.add_parser("plan")
    backup = sub.add_parser("backup")
    backup.add_argument("target")
    args = parser.parse_args()
    con = None
    try:
        if args.command == "add" and (not args.request or not args.category or not 0 < args.cents <= 2**63 - 1):
            raise ValueError("request/category required; cents must be a positive signed 64-bit integer")
        con = connect(args.db, "rwc" if args.command == "init" else "ro" if args.command in ("list", "plan", "backup") else "rw")
        if args.command == "init":
            mode = con.execute("PRAGMA journal_mode=WAL").fetchone()[0]
            if mode != "wal":
                raise ValueError("WAL unavailable")
            con.execute("CREATE TABLE IF NOT EXISTS expenses(request TEXT PRIMARY KEY NOT NULL, category TEXT NOT NULL, cents INTEGER NOT NULL CHECK(cents>0)) STRICT")
            con.execute("CREATE INDEX IF NOT EXISTS expenses_category_request ON expenses(category,request)")
            result = {"initialized": True}
        elif args.command == "add":
            con.execute("BEGIN IMMEDIATE")
            old = con.execute("SELECT category,cents FROM expenses WHERE request=?", (args.request,)).fetchone()
            if old is not None and old != (args.category, args.cents):
                raise ValueError("request reused with different payload")
            if old is None:
                con.execute("INSERT INTO expenses VALUES(?,?,?)", (args.request, args.category, args.cents))
            con.execute("COMMIT")
            result = {"request": args.request, "replayed": old is not None}
        elif args.command == "list":
            result = con.execute("SELECT category,SUM(cents),COUNT(*) FROM expenses GROUP BY category ORDER BY category").fetchall()
        elif args.command == "plan":
            result = con.execute("EXPLAIN QUERY PLAN SELECT request FROM expenses WHERE category=? ORDER BY request", ("food",)).fetchall()
        else:
            target = Path(args.target).resolve()
            if target == Path(args.db).resolve():
                raise ValueError("backup target must differ from source")
            # Exclusive creation prevents replacing a previous backup.
            with target.open("xb"):
                pass
            dest = None
            try:
                dest = connect(target, "rw")
                con.backup(dest)
                if dest.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
                    raise ValueError("backup integrity failed; keep target for inspection")
            finally:
                if dest is not None:
                    dest.close()
            result = {"backup": str(target)}
        print(json.dumps(result, ensure_ascii=False))
        return 0
    except (sqlite3.Error, OSError, ValueError) as exc:
        if con is not None and con.in_transaction:
            con.execute("ROLLBACK")
        print(str(exc), file=sys.stderr)
        return 1
    finally:
        if con is not None:
            con.close()


if __name__ == "__main__":
    sys.exit(main())
txt
# CPython 3.12+ standard library only; sqlite3 runtime >=3.37 for STRICT.
# No pip packages. Example was not executed in this delivery.

备份代码先以 xb 创建目标，防止覆盖既有备份，然后通过 Connection.backup 取得数据库快照。若备份失败，会留下待检查的目标文件，操作者应确认原因后自行处理；代码不擅自删除文件。使用 readonly 源连接并不意味着不需要原 WAL 可读条件。

错误、退出码与阅读验证 ​

SQLite 退出码契约是成功为 0，文件、数据库和业务错误为 1，argparse 参数格式错误为 2。现有目标备份触发 FileExistsError；不存在源库触发 unable to open database file；竞争写入可能触发 database is locked。标准错误承载说明，标准输出仅承载成功 JSON。底层 rollback 失败等极端 I/O 故障可能中断清理，不能据此承诺数据盘故障可自动恢复。

静态检查只证明 Python 语法可解析，不证明 SQL、文件权限、锁等待、备份或输出成功。本轮这些均未执行。读者后续验证应使用独立临时目录，依次检查新请求、同参数重放、异参数拒绝、非法 cents、未知文件、只读权限和独立备份查询，再记录实际版本和退出码。

综合练习与参考方案 ​

练习：增加按 category 列出明细的子命令，要求不拼 SQL、不丢失稳定顺序、不接受任意 SQL 字符串。先写出预期：food 只返回一条 r1，而缺失分类返回空数组。

参考方案：新增固定 SELECT request,cents，WHERE category=?，ORDER BY request；通过参数对象传 category。沿用 ro 连接和现有错误契约。此处不额外实现功能，目的是让读者独立完成并说明索引为何适配。

继续阅读