SQLite：给程序加一个数据库

程序的数据一直存在文件里，够用吗？文件适合小数据、单机、自己看；一旦数据多了、要查询、要并发，就该上数据库了。这一篇学 SQLite——Python 内置、零配置、一个文件就是一个库，个人项目和工具的首选。

为什么需要数据库

对比一下文件方案和数据库方案：

• 文件：读要全量加载，改要重写整个文件，并发写会坏，没有查询语言
• 数据库：按条件查（”上个月花了多少钱”）、只改要改的行、支持并发、数据量大也不怕

SQLite 是嵌入式数据库：一个 .db 文件就是整个数据库，不需要安装服务，Python 内置 sqlite3 模块直接用。个人项目、小工具、网站原型，它都够用。

连接和建表

import sqlite3

# 连接（文件不存在会自动创建）
conn = sqlite3.connect("ledger.db")
cur = conn.cursor()

# 建表：CREATE TABLE IF NOT EXISTS 防止重复建
cur.execute("""
CREATE TABLE IF NOT EXISTS records (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    category TEXT NOT NULL,
    amount REAL NOT NULL,
    note TEXT,
    created_at TEXT DEFAULT (datetime('now', 'localtime'))
)
""")
conn.commit()   # 提交！不提交不生效
conn.close()    # 用完关闭

SQL 语法先认识这几个：CREATE TABLE 建表、INTEGER PRIMARY KEY AUTOINCREMENT 自增主键、TEXT/REAL 类型、NOT NULL 非空约束、DEFAULT 默认值。

增：INSERT

import sqlite3

conn = sqlite3.connect("ledger.db")
cur = conn.cursor()

# 方式 1：直接拼 SQL（危险！会被 SQL 注入，别这么写）
# cur.execute(f"INSERT INTO records (category, amount) VALUES ('{cat}', {amt})")

# 方式 2：参数化（正确！用 ? 占位，值作为参数传入）
cur.execute(
    "INSERT INTO records (category, amount, note) VALUES (?, ?, ?)",
    ("吃饭", 12.5, "午饭"),
)
cur.execute(
    "INSERT INTO records (category, amount, note) VALUES (?, ?, ?)",
    ("交通", 5.0, "公交"),
)
conn.commit()
print("插入成功，最后一条 id：", cur.lastrowid)
conn.close()

永远用参数化查询（? 占位），永远不要拼字符串。拼字符串会被 SQL 注入——用户输入 '; DROP TABLE records;-- 这种内容，你的表就没了。这是安全红线。

查：SELECT

conn = sqlite3.connect("ledger.db")
cur = conn.cursor()

# 查全部
cur.execute("SELECT * FROM records")
rows = cur.fetchall()          # 列表套元组
for row in rows:
    print(row)                 # (1, '吃饭', 12.5, '午饭', '2026-08-14 16:30:00')

# 带条件查询
cur.execute("SELECT * FROM records WHERE category = ?", ("吃饭",))
rows = cur.fetchall()

# 聚合：统计
cur.execute("SELECT category, SUM(amount) FROM records GROUP BY category")
for category, total in cur.fetchall():
    print(f"{category}: {total}")

# 排序和限制
cur.execute("SELECT * FROM records ORDER BY amount DESC LIMIT 3")   # 花最多的 3 条

# 取单条
cur.execute("SELECT * FROM records WHERE id = ?", (1,))
row = cur.fetchone()
print(row)

conn.close()

fetchall 取全部、fetchone 取一条。GROUP BY 分组统计、ORDER BY ... DESC 排序、LIMIT n 限量——这几个组合起来能回答大部分查询需求。

改和删：UPDATE / DELETE

conn = sqlite3.connect("ledger.db")
cur = conn.cursor()

# 改：更新 id=1 的备注
cur.execute("UPDATE records SET note = ? WHERE id = ?", ("午饭（食堂）", 1))
print("修改行数：", cur.rowcount)    # 影响了几行

# 删：删除指定 id
cur.execute("DELETE FROM records WHERE id = ?", (3,))

# 删全部（危险！）
# cur.execute("DELETE FROM records")

conn.commit()
conn.close()

注意 WHERE 条件——UPDATE/DELETE 不带 WHERE 会作用到所有行。写删除语句前先想三遍。

事务：要么全成功，要么全不成功

转账这种操作必须”原子”：扣钱和加钱要么都成功，要么都失败。SQLite 默认开启事务：

conn = sqlite3.connect("ledger.db")
cur = conn.cursor()
# 先建一个账户表（示例用）
cur.execute("CREATE TABLE IF NOT EXISTS accounts (name TEXT PRIMARY KEY, balance REAL)")
cur.execute("INSERT OR IGNORE INTO accounts VALUES ('小明', 1000), ('小红', 500)")
try:
    cur.execute("UPDATE accounts SET balance = balance - 100 WHERE name = ?", ("小明",))
    cur.execute("UPDATE accounts SET balance = balance + 100 WHERE name = ?", ("小红",))
    conn.commit()          # 都成功才提交
except Exception:
    conn.rollback()        # 出任何错都回滚，数据回到操作前
    print("操作失败，已回滚")
finally:
    conn.close()

commit 提交、rollback 回滚。用 with conn: 也行——with 块正常结束自动 commit，异常自动 rollback。

实战：把记账本升级到数据库版

import sqlite3

class LedgerDB:
    def __init__(self, path="ledger.db"):
        self.conn = sqlite3.connect(path)
        self.cur = self.conn.cursor()
        self.cur.execute("""
            CREATE TABLE IF NOT EXISTS records (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                category TEXT NOT NULL,
                amount REAL NOT NULL,
                note TEXT
            )
        """)
        self.conn.commit()

    def add(self, category, amount, note=""):
        self.cur.execute(
            "INSERT INTO records (category, amount, note) VALUES (?, ?, ?)",
            (category, amount, note),
        )
        self.conn.commit()

    def total(self):
        self.cur.execute("SELECT SUM(amount) FROM records")
        return self.cur.fetchone()[0] or 0

    def by_category(self):
        self.cur.execute("SELECT category, SUM(amount) FROM records GROUP BY category")
        return self.cur.fetchall()

    def close(self):
        self.conn.close()

db = LedgerDB()
db.add("吃饭", 12.5, "午饭")
db.add("吃饭", 8.0, "晚饭")
db.add("交通", 5.0)
print("总支出：", db.total())            # 25.5
print("分类统计：", db.by_category())    # [('吃饭', 20.5), ('交通', 5.0)]
db.close()

什么时候换 MySQL / PostgreSQL

SQLite 的局限：并发写能力弱、没有用户权限体系、不适合多机部署。出现这些信号再升级：

• 多个进程/服务器同时高频写入
• 数据量上亿级别
• 需要复杂的用户权限管理
• 要主从复制、高可用

升级后 SQL 基本不变（都是标准 SQL），主要换连接方式。

新手坑

坑 1：忘 commit。INSERT/UPDATE/DELETE 后不 commit，关连接数据就没了。或者用 with conn: 自动提交。

坑 2：SQL 注入。拼字符串查库是最大的坑。一律参数化 ?。

坑 3：忘 close。连接不关会占用文件锁。用完 close，或用 with 上下文管理。

小结

• SQLite：内置、零配置、单文件，个人项目首选；sqlite3.connect 即用
• 四类操作：INSERT 增、SELECT 查（WHERE/GROUP BY/ORDER BY/LIMIT）、UPDATE 改、DELETE 删
• 参数化查询防 SQL 注入，永远别拼字符串
• 事务：commit 提交、rollback 回滚，with conn: 自动处理
• 并发写大、要权限管理时再考虑 MySQL/PostgreSQL

练习

1. 建一个”图书”表（书名、作者、价格、读完没），插入 5 本，查询价格大于 50 的
2. 写一个函数 search_records(keyword)，按备注模糊搜索（提示：LIKE ‘%’ || ? || ‘%’）
3. 思考：为什么说”DELETE FROM records”不带 WHERE 很危险？结合事务讲一下：如果误删了全表数据，有什么补救手段？