高级篇 / 第 17 章

第 17 章 · 数据库编程实战

为什么需要数据库：文件存储的边界

到目前为止，我们用 JSON 文件存过数据、用 pickle 序列化过对象。这些方式在小规模场景下足够，但一旦数据量上到几万行、或者有多个用户同时写入，边界就暴露出来了：

1. 没有事务：写文件的过程如果中途崩溃，可能得到半个文件；两个进程同时写会互相覆盖。
2. 没有并发控制：没有行锁、表锁，读到的是别的进程写了一半的数据。
3. 没有索引：每次查询都要全表扫描，几万行时响应从毫秒退化到秒级。
4. 没有约束：主键重复、外键指向不存在的行、邮箱字段存了两个不同写法，全靠你自己在代码里 if 判断。
5. 没有并发访问协议：没有网络层的认证、权限、连接管理。

数据库管理系统（DBMS, Database Management System）就是为解决这些问题而生的。数据库（database）是组织数据的集合，DBMS 是管理它的软件，如 PostgreSQL、MySQL、SQLite。

关系模型与非关系模型

关系模型（relational model）由 E. F. Codd 于 1970 年提出，核心思想是：把数据存成一张张二维表（relation），表由行和列组成，行叫记录/元组（tuple/row），列叫字段/属性（attribute/column），表之间通过键（key）关联。用表和键来组织数据，并通过 SQL 语言操作的数据库叫关系型数据库（RDBMS）。

非关系型数据库（NoSQL）泛指不遵循关系模型的数据库，常见类别有：

SQL 入门：增删改查四板斧

SQL（Structured Query Language，结构化查询语言）是操作关系数据库的标准语言。语法关键字不区分大小写，通常统一写成大写。

DDL：定义结构

-- 建表：主键、自增、非空、唯一、默认值
CREATE TABLE users (
    id         INTEGER PRIMARY KEY AUTOINCREMENT,
    username   TEXT    NOT NULL UNIQUE,
    email      TEXT    NOT NULL UNIQUE,
    age        INTEGER CHECK (age >= 0 AND age <= 150),
    bio        TEXT    DEFAULT '',
    created_at TEXT    NOT NULL DEFAULT (datetime('now'))
);

-- 索引：为高频查询字段建索引
CREATE INDEX idx_users_age ON users (age);

-- 改表结构
ALTER TABLE users ADD COLUMN city TEXT DEFAULT '';

-- 看结构 / 删表
.schema users
DROP TABLE users;

DML：操作数据

-- 插入：单行、多行、返回自增主键
INSERT INTO users (username, email, age)
VALUES ('alice', 'alice@example.com', 28);

INSERT INTO users (username, email, age) VALUES
    ('bob',   'bob@example.com',   31),
    ('carol', 'carol@example.com', 24);

-- 查询：条件、排序、分页、聚合
SELECT id, username, age FROM users WHERE age > 25 ORDER BY age DESC LIMIT 10 OFFSET 0;

SELECT COUNT(*) AS total, AVG(age) AS avg_age, MAX(age) AS oldest
FROM users;

SELECT username, age FROM users
WHERE email LIKE '%@example.com'      -- 通配符 % 匹配任意多字符，_ 匹配一个字符
ORDER BY created_at
LIMIT 10;

-- 聚合 + 分组 + HAVING（对分组后的结果过滤）
SELECT age / 10 * 10 AS age_group, COUNT(*) AS cnt
FROM users
GROUP BY age_group
HAVING cnt >= 2;

-- 更新与删除（务必带 WHERE！）
UPDATE users SET age = age + 1 WHERE username = 'alice';
DELETE FROM users WHERE age < 18;

-- UPSERT：有则更新，无则插入
INSERT INTO users (username, email, age) VALUES ('alice', 'new@example.com', 29)
ON CONFLICT(username) DO UPDATE SET email = excluded.email, age = excluded.age;

JOIN：把表拼起来

-- 一对多：查询每个用户及其所有文章
SELECT u.username, p.title, p.published_at
FROM users u
INNER JOIN posts p ON p.user_id = u.id
ORDER BY p.published_at DESC;

-- 左连接：列出所有用户，包括没发过文章的人
SELECT u.username, COUNT(p.id) AS post_count
FROM users u
LEFT JOIN posts p ON p.user_id = u.id
GROUP BY u.id
HAVING post_count = 0;

sqlite3：标准库内置数据库

SQLite 是一个嵌入式（embedded）数据库：整个数据库就是一个文件，不需要安装服务，不需要独立进程，Python 标准库自带 sqlite3 模块。它是教学、小型工具、桌面应用、移动 App 的绝佳选择，也是本章所有示例的运行环境。

最小可运行示例

demo.py —— 建库建表并插入数据

import sqlite3
from pathlib import Path

DB_FILE = Path("shop.db")

# 连接：文件不存在会自动创建
conn = sqlite3.connect(DB_FILE)
cur = conn.cursor()   # 游标（cursor）：执行 SQL 并取回结果的对象

cur.execute("""
    CREATE TABLE IF NOT EXISTS products (
        id    INTEGER PRIMARY KEY AUTOINCREMENT,
        name  TEXT NOT NULL UNIQUE,
        price REAL NOT NULL CHECK (price >= 0),
        stock INTEGER NOT NULL DEFAULT 0
    )
""")

cur.executemany(
    "INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
    [("机械键盘", 399.0, 12), ("人体工学椅", 1299.0, 3), ("显示器", 1599.0, 7)],
)

conn.commit()   # 提交事务，把内存中的改动写入磁盘
conn.close()

print("数据库文件：", DB_FILE.resolve(), "大小：", DB_FILE.stat().st_size, "字节")

预期输出：

数据库文件： /你的目录/shop.db 大小： 16384 字节

游标的四种取数方式

demo.py —— 游标 API 对比

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE t (id INTEGER, name TEXT, score REAL)")
conn.executemany(
    "INSERT INTO t VALUES (?, ?, ?)",
    [(1, "alice", 88.5), (2, "bob", 92.0), (3, "carol", 71.5)],
)
conn.commit()

cur = conn.cursor()

# 1) fetchone：取一行，取完游标停在下一行
cur.execute("SELECT id, name, score FROM t ORDER BY id")
print(cur.fetchone())    # (1, 'alice', 88.5)
print(cur.fetchone())    # (2, 'bob', 92.0)

# 2) fetchmany(n)：取 n 行
cur.execute("SELECT id FROM t ORDER BY id")
print(cur.fetchmany(2))  # [(1,), (2,)]

# 3) fetchall：取剩下全部
cur.execute("SELECT id FROM t ORDER BY id")
print(cur.fetchall())    # [(3,)]

# 4) 一步到位：conn.execute() 内部就帮你建了游标
rows = conn.execute("SELECT name, score FROM t WHERE score > ? ORDER BY score DESC", (80,)).fetchall()
print(rows)              # [('bob', 92.0), ('alice', 88.5)]

# 遍历：遍历本身是惰性的，注意大结果集别用 fetchall
for (name, score) in conn.execute("SELECT name, score FROM t"):
    print(f"{name:6s} {score:6.1f}")

conn.close()

row_factory：让结果变成字典

import sqlite3

conn = sqlite3.connect(":memory:", row_factory=sqlite3.Row)
conn.execute("CREATE TABLE t (id INTEGER, name TEXT)")
conn.execute("INSERT INTO t VALUES (1, 'alice')")
conn.commit()

row = conn.execute("SELECT * FROM t").fetchone()
print(row["name"])      # alice
print(dict(row))        # {'id': 1, 'name': 'alice'}
print(row.keys())       # ['id', 'name']

# SQLAlchemy 中常用的 dict 工厂
conn.row_factory = sqlite3.Row
conn.close()

参数化查询：SQL 注入的第一道防线

把用户输入直接拼进 SQL，是最常见也最危险的错误。所谓 SQL 注入（SQL injection），攻击者通过构造输入改变 SQL 语义，从而绕过校验、窃取或篡改数据。

错误写法 vs 正确写法

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE users (id INTEGER, username TEXT, password TEXT)")
conn.executemany(
    "INSERT INTO users VALUES (?, ?, ?)",
    [(1, "admin", "secret"), (2, "alice", "pw123")],
)
conn.commit()

# ❌ 危险：字符串拼接，注入点遍地
name = input_name = "admin' OR '1'='1"
bad_sql = f"SELECT * FROM users WHERE username = '{input_name}'"
# bad_sql 变成：SELECT * FROM users WHERE username = 'admin' OR '1'='1'
# 结果：返回了所有用户的密码
print(conn.execute(bad_sql).fetchall())

# ✅ 正确：占位符 + 参数元组，SQL 结构与数据分离
good_sql = "SELECT * FROM users WHERE username = ?"
print(conn.execute(good_sql, (input_name,)).fetchall())  # []

# ✅ 更安全的现代写法：命名参数，顺序无关、可读性好
sql = "SELECT * FROM users WHERE username = :name AND password = :pwd"
print(conn.execute(sql, {"name": "alice", "pwd": "pw123"}).fetchall())

# ✅ 插入同样要用占位符
conn.execute("INSERT INTO users (username, password) VALUES (?, ?)", ("dave", "pw456"))
conn.commit()
conn.close()

# IN 子句的正确处理：占位符个数必须和参数个数一致
ids = [3, 7, 11]
placeholders = ", ".join("?" * len(ids))
sql = f"SELECT id, name FROM users WHERE id IN ({placeholders})"
rows = conn.execute(sql, ids).fetchall()
# 注意：这里 f-string 只用于生成 ?? 占位符本身，用户数据仍然走参数

# LIKE 的转义
def like_escape(text: str) -> str:
    return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

sql = "SELECT * FROM users WHERE username LIKE ? ESCAPE '\\'"
rows = conn.execute(sql, (f"%{like_escape('100%_real')}%",)).fetchall()

事务与 ACID

事务（transaction）是一组要么全部成功、要么全部失败的操作集合。ACID 是它的四个性质：

Python 里手动控制事务

demo.py —— 显式开启、提交、回滚

import sqlite3

conn = sqlite3.connect("bank.db", isolation_level=None)  # None = 自动提交，改为手动控制
cur = conn.cursor()
cur.execute("CREATE TABLE IF NOT EXISTS accounts (name TEXT PRIMARY KEY, balance REAL NOT NULL)")

def transfer(cur, frm: str, to: str, amount: float) -> bool:
    try:
        cur.execute("UPDATE accounts SET balance = balance - ? WHERE name = ? AND balance >= ?",
                    (amount, frm, amount))
        if cur.rowcount == 0:
            print(f"扣款失败：{frm} 余额不足或不存在")
            cur.execute("ROLLBACK")
            return False
        cur.execute("UPDATE accounts SET balance = balance + ? WHERE name = ?", (amount, to))
        cur.execute("COMMIT")
        print(f"转账成功 {amount} 元：{frm} -> {to}")
        return True
    except sqlite3.Error as e:
        cur.execute("ROLLBACK")
        print("事务已回滚：", e)
        return False

cur.execute("INSERT OR IGNORE INTO accounts VALUES ('alice', 1000.0)")
cur.execute("INSERT OR IGNORE INTO accounts VALUES ('bob', 0.0)")

print(transfer(cur, "alice", "bob", 300.0))
print(transfer(cur, "alice", "bob", 99999.0))   # 余额不足，回滚

for row in cur.execute("SELECT name, balance FROM accounts"):
    print(row)

cur.close()
conn.close()

预期输出：

转账成功 300.0 元：alice -> bob
扣款失败：alice 余额不足或不存在
('alice', 700.0)
('bob', 300.0)

隔离级别

并发访问：线程安全、锁与连接池

单线程脚本随手 create_connection 没问题，但一旦进入 Web 服务或线程池，就有两个必须解决的问题：sqlite3 连接默认不能跨线程使用，以及连接创建开销大。

线程本地连接模式

import sqlite3
import threading

DB = "shop.db"
local = threading.local()   # 线程本地存储：每个线程一份独立数据

def get_conn() -> sqlite3.Connection:
    conn = getattr(local, "conn", None)
    if conn is None:
        conn = sqlite3.connect(DB, check_same_thread=False)
        conn.row_factory = sqlite3.Row
        local.conn = conn
    return conn

results = []

def worker(n: int) -> None:
    conn = get_conn()
    conn.execute("INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
                 (f"item-{n}", n * 10.0, n))
    conn.commit()
    results.append(conn.execute("SELECT COUNT(*) FROM products").fetchone()[0])

threads = [threading.Thread(target=worker, args=(i,)) for i in range(1, 6)]
for t in threads:
    t.start()
for t in threads:
    t.join()

conn = get_conn()
print("总行数：", conn.execute("SELECT COUNT(*) FROM products").fetchone()[0])

连接池：把「建连接」变成可复用资源

生产数据库（如 PostgreSQL、MySQL）建立 TCP 连接 + 认证握手通常要几毫秒到几十毫秒，高并发下自建自销会成为瓶颈。连接池（connection pool）预先建立一批连接并循环借还。Python 里最常用的是 queue.Queue 手写实现，或直接用 DBUtils.PooledDB / sqlalchemy.pool。

import sqlite3
import threading
import queue
from contextlib import contextmanager

class SQLitePool:
    """极简连接池：固定数量连接，用完还回。"""

    def __init__(self, db: str, size: int = 5) -> None:
        self._pool: queue.Queue[sqlite3.Connection] = queue.Queue(maxsize=size)
        for _ in range(size):
            conn = sqlite3.connect(db, check_same_thread=False)
            conn.row_factory = sqlite3.Row
            self._pool.put(conn)

    @contextmanager
    def acquire(self):
        """借出连接，用完自动归还（含异常）。"""
        conn = self._pool.get()
        try:
            yield conn
        finally:
            self._pool.put(conn)


def export_csv(pool: SQLitePool, path: str) -> int:
    with pool.acquire() as conn:
        rows = conn.execute("SELECT id, name, price FROM products ORDER BY id").fetchall()
    with open(path, "w", encoding="utf-8") as f:
        f.write("id,name,price\n")
        for r in rows:
            f.write(f"{r['id']},{r['name']},{r['price']}\n")
    return len(rows)


# 并发导出演示
errs = []
def task(i: int) -> None:
    try:
        export_csv(pool, f"out_{i}.csv")
    except Exception as e:   # noqa: BLE001
        errs.append(e)

pool = SQLitePool("shop.db", size=3)
ts = [threading.Thread(target=task, args=(i,)) for i in range(10)]
for t in ts:
    t.start()
for t in ts:
    t.join()
print("完成，错误数：", len(errs))

conn = sqlite3.connect("shop.db")
conn.execute("PRAGMA journal_mode = WAL")        # 写日志模式，读写并发
conn.execute("PRAGMA synchronous = NORMAL")     # 略降持久性，换更高吞吐
conn.execute("PRAGMA foreign_keys = ON")        # SQLite 默认不开外键！必须手动开
conn.execute("PRAGMA busy_timeout = 5000")      # 锁等待 5 秒再报 database is locked
conn.commit()

SQLAlchemy ORM：可维护的数据层

ORM（Object-Relational Mapping，对象关系映射）把「表的一行」映射为「Python 的一个对象」，让你用类和方法操作数据库，而不是手写 SQL。SQLAlchemy 是 Python 生态最主流的 ORM，其 2.0 版采用 ORM 2.0 风格，类型提示更严格。

安装与声明模型

pip install "sqlalchemy>=2.0" 

models.py —— 声明两张表并建立关系

from __future__ import annotations

from datetime import datetime

from sqlalchemy import ForeignKey, String, Integer, DateTime, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    """所有 ORM 模型的基类。"""


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(120), unique=True)
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    # 一对多：一个用户有多篇文章
    posts: Mapped[list[Post]] = relationship(back_populates="author", cascade="all, delete-orphan")


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text, default="")
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)

    author: Mapped[User] = relationship(back_populates="posts")

app.py —— 创建引擎、会话与基础 CRUD

from datetime import datetime

from sqlalchemy import create_engine, select, func, desc
from sqlalchemy.orm import Session, sessionmaker

from models import Base, User, Post

# 引擎（engine）：管理连接池。echo=True 会打印所有 SQL，调试神器。
engine = create_engine("sqlite:///app.db", echo=False, future=True)

# create_all 只创建不存在的表，不会修改已有表结构
Base.metadata.create_all(engine)

SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)


def create_user(session: Session, username: str, email: str) -> User:
    user = User(username=username, email=email)
    session.add(user)                 # 加入会话（此时还没写库）
    session.commit()                  # 提交：此时才真正 INSERT
    session.refresh(user)             # 刷新，拿到数据库生成的字段（如自增 id）
    return user


def get_user(session: Session, username: str) -> User | None:
    stmt = select(User).where(User.username == username)
    return session.scalar(stmt)


def list_active_users(session: Session, limit: int = 10) -> list[User]:
    stmt = select(User).where(User.is_active.is_(True)).order_by(desc(User.created_at)).limit(limit)
    return list(session.scalars(stmt))


def add_post(session: Session, author: User, title: str, content: str = "") -> Post:
    post = Post(title=title, content=content, author=author)
    session.add(post)
    session.commit()
    session.refresh(post)
    return post


# ---- 运行 ----
with SessionLocal() as session:
    alice = create_user(session, "alice", "alice@example.com")
    add_post(session, alice, "第一篇", "正文内容")
    add_post(session, alice, "第二篇")

    for u in list_active_users(session):
        print(u.id, u.username, u.created_at)

    # 统计：func.count + group_by
    stmt = (
        select(User.username, func.count(Post.id).label("n"))
        .join(Post, Post.author_id == User.id)
        .group_by(User.id)
        .having(func.count(Post.id) >= 1)
    )
    for name, n in session.execute(stmt):
        print(f"{name} 发了 {n} 篇")

增删改查 API 速查

# 更新与删除
with SessionLocal() as session:
    user = get_user(session, "alice")
    if user:
        user.email = "alice@new.com"
        user.is_active = False
        session.commit()

    # 批量删除：先查出对象再删（ORM 风格）
    for post in session.scalars(select(Post).where(Post.title == "草稿")):
        session.delete(post)
    session.commit()

    # 关联对象：user.posts 是列表，遍历时自动发出懒加载 SQL
    user = get_user(session, "alice")
    if user:
        for p in user.posts:
            print(p.title)

# 复杂条件组合
from sqlalchemy import or_, and_, not_
stmt = select(User).where(
    or_(User.username.like("a%"), and_(User.age < 20, User.city == "上海")),
    not_(User.is_active.is_(False)),
)
# 查看最终生成的 SQL
from sqlalchemy.dialects import sqlite
print(stmt.compile(dialect=sqlite.dialect()))

惰性加载陷阱与 N+1 问题

from sqlalchemy.orm import selectinload

# ❌ N+1
for u in session.scalars(select(User)):
    print(u.username, len(u.posts))     # 每次访问 posts 都发一条 SQL

# ✅ 预加载：只用 2 条 SQL（1 条查 user，1 条查全部 post）
users = session.scalars(select(User).options(selectinload(User.posts))).all()
for u in users:
    print(u.username, len(u.posts))     # 不再发 SQL

# ✅ 或者直接 JOIN 联表查询
stmt = select(User.username, Post.title).join(Post, Post.author_id == User.id)
for uname, title in session.execute(stmt):
    print(uname, "-", title)

换 PostgreSQL 只需改一行

# pip install "psycopg[binary]"
engine = create_engine("postgresql+psycopg://user:pass@localhost:5432/mydb", pool_size=5, max_overflow=10)
# 其他代码一行不用改 —— 这就是 ORM 的价值

Redis 缓存：给数据库加一层加速

缓存（cache）的思路是：把读取频繁、计算昂贵或变更很少的数据临时放在更快的存储里。Redis 是一个内存数据结构服务器（in-memory data structure store），因为数据在内存中，读写通常是微秒级，比磁盘数据库快一到两个数量级。

基础用法

# pip install redis
import redis

# decode_responses=True 让返回值变成 str 而不是 bytes
r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

r.set("user:1:name", "alice")
print(r.get("user:1:name"))          # alice
print(r.exists("user:1:name"))        # 1
print(r.ttl("user:1:name"))           # -1（永不过期），-2（不存在）

r.setex("user:2:name", 60, "bob")    # 设置并带 60 秒过期
r.hset("user:1", mapping={"name": "alice", "email": "a@x.com", "age": 28})
print(r.hgetall("user:1"))           # {'name': 'alice', 'email': 'a@x.com', 'age': '28'}

r.sadd("tags:python", "flask", "sqlalchemy")
print(r.smembers("tags:python"))

r.zadd("rank:score", {"alice": 100, "bob": 80})
print(r.zrevrange("rank:score", 0, 1, withscores=True))

缓存旁路模式（Caching Aside）

最常用的策略：读的时候先查缓存，没有再查数据库并写回缓存；写的时候先更新数据库，然后删除缓存（而不是更新缓存）。

import json
import sqlite3
import redis
from typing import Any

r = redis.Redis(decode_responses=True)
db = sqlite3.connect("shop.db")
db.row_factory = sqlite3.Row

def get_product(pid: int) -> dict[str, Any] | None:
    """读：缓存 -> 数据库 -> 回填缓存"""
    key = f"product:{pid}"
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)          # 命中缓存
    row = db.execute("SELECT id, name, price, stock FROM products WHERE id = ?", (pid,)).fetchone()
    if row is None:
        return None
    data = dict(row)
    r.setex(key, 300, json.dumps(data))   # TTL 300 秒兜底，防止脏数据永久存在
    return data


def update_stock(pid: int, delta: int) -> None:
    """写：先改数据库，再删缓存（缓存失效而非更新）"""
    db.execute("UPDATE products SET stock = stock + ? WHERE id = ?", (delta, pid))
    db.commit()
    r.delete(f"product:{pid}")


print(get_product(1))     # 第一次：查数据库 + 写缓存
print(get_product(1))     # 第二次：命中缓存
update_stock(1, -1)       # 扣库存，缓存失效
print(get_product(1))     # 再次：重新查库并回填

缓存三大经典问题

import time, random, threading

# 空值缓存防穿透
MISS = object()

def get_product_safe(pid: int) -> Any:
    key = f"product:{pid}"
    cached = r.get(key)
    if cached == "__none__":
        return None                        # 记住「确实不存在」
    if cached is not None:
        return json.loads(cached)
    row = db.execute("SELECT id, name, price, stock FROM products WHERE id = ?", (pid,)).fetchone()
    if row is None:
        r.setex(key, 60, "__none__")       # 短 TTL 占位
        return None
    data = dict(row)
    r.setex(key, 300 + random.randint(0, 60), json.dumps(data))  # TTL 抖动防雪崩
    return data


# 互斥锁防击穿（用 SETNX 模拟）
def get_product_with_lock(pid: int) -> Any:
    key = f"product:{pid}"
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)

    lock_key = f"lock:product:{pid}"
    got_lock = r.set(lock_key, "1", nx=True, ex=5)   # 只让一个线程回源
    if got_lock:
        try:
            row = db.execute("SELECT * FROM products WHERE id = ?", (pid,)).fetchone()
            if row:
                r.setex(key, 300, json.dumps(dict(row)))
                return dict(row)
        finally:
            r.delete(lock_key)
    else:
        time.sleep(0.05)                  # 没抢到锁，等一下重试读缓存
        cached = r.get(key)
        if cached:
            return json.loads(cached)
    return None

工程实践清单

1. 永远用参数化查询。任何 f-string 拼 SQL 都要在 code review 被打回。
2. 连接生命周期要明确：Web 应用一个请求一个连接，用完即还；不要把连接当全局变量长期持有。
3. 事务边界放在业务层，不要散落在 DAO 里，否则无法保证「扣库存 + 建订单」的原子性。
4. 给数据库加索引，但索引不是越多越好：每个索引都会拖慢写入。先用 EXPLAIN 分析执行计划。
5. 敏感信息不落日志。打印 SQL 时注意密码、身份证号等字段脱敏。
6. 迁移脚本管理表结构。用 Alembic（SQLAlchemy 官方迁移工具）而不是手动 ALTER TABLE。
7. 备份与恢复演练。没有验证过恢复流程的备份等于没有备份。

-- 用 EXPLAIN 看查询计划：搜索全表扫描（SCAN）就该加索引了
EXPLAIN QUERY PLAN SELECT * FROM products WHERE name = '机械键盘';

-- 输出示例：
-- SCAN TABLE products          <-- 没有索引，全表扫描
-- SEARCH TABLE products USING INDEX idx_products_name (name=?)   <-- 走索引了

本章小结

• 关系模型用二维表和键组织数据；RDBMS 与 NoSQL 各有适用场景，真实系统常混用。
• SQL 四板斧：INSERT / SELECT / UPDATE / DELETE，配上 JOIN、GROUP BY、HAVING 完成复杂查询。
• sqlite3 是零依赖的嵌入式数据库，connect → cursor → execute → commit → close 五步走；fetchone/fetchmany/fetchall 决定取数方式。
• 参数化查询（? 占位符或命名参数）是防 SQL 注入的第一道防线；标识符（表名列名）必须走白名单。
• ACID 与事务：用 try/except 包裹业务逻辑，失败时 ROLLBACK，保证不产生半截数据。
• 并发：sqlite3 连接不能跨线程共享，用 threading.local 或连接池；开启 WAL 提升读写并发。
• SQLAlchemy 2.0 ORM：create_engine → sessionmaker → Session，select() + session.scalars() 是 2.0 标准查询写法；注意 N+1 问题，用 selectinload 预加载。
• Redis 缓存用缓存旁路模式：读时回填、写时删除；穿透用空值缓存、击穿用互斥锁、雪崩用 TTL 抖动。

练习题

1. 建库建表与批量插入：用 sqlite3 创建一个 library.db，含 books（id、title、author、price、stock）与 borrowers（id、name、email）两张表，用 executemany 各插入 10 条数据，并打印表结构。
   参考答案要点：conn.execute("CREATE TABLE ...") 建表；conn.executemany(sql, [(...), ...]) 批量插入；最后 conn.commit()。用 conn.execute("SELECT sql FROM sqlite_master") 可查看建表语句。
2. 防注入改造：下面这段代码有严重安全问题，请指出所有注入点并改写为参数化查询。提示：username、age、sort_by 三个输入都来自用户。
   参考答案要点：f"SELECT * FROM users WHERE username='{name}' AND age > {age} ORDER BY {sort_by}" 中前两个改用 ? 占位符；sort_by 因为是标识符无法用占位符，必须白名单：sort_col = {"name": "username", "age": "age"}.get(sort_by, "username")，不在字典里就用默认值。
3. 事务与回滚：实现 transfer(conn, frm, to, amount)，从 frm 转 amount 给 to；若 frm 余额不足则打印提示并回滚，不产生任何改动。测试三种场景：正常转账、余额不足、账户不存在。
   参考答案要点：先查 frm 余额（或在 UPDATE 的 WHERE 里加 balance >= ? 并检查 cursor.rowcount == 0），异常时执行 conn.rollback()。用完打印两账户余额验证总和不变量。
4. 聚合统计查询：用一条 SQL 统计每个作者的图书数量和平均定价，只保留数量 ≥ 2 的作者，按数量降序排列。
   参考答案代码：SELECT author, COUNT(*) AS cnt, AVG(price) AS avg_price FROM books GROUP BY author HAVING COUNT(*) >= 2 ORDER BY cnt DESC。注意 HAVING 里可以用 COUNT(*) 别名 cnt（SQLite 支持，部分数据库不支持，稳妥起见写完整表达式）。
5. JOIN 查询：LEFT JOIN 查出所有作者及其图书数，包括没有图书的作者（结果应为 0）。
   参考答案代码：SELECT b.author, COUNT(b.id) AS n FROM authors a LEFT JOIN books b ON b.author = a.author GROUP BY a.author。注意必须 COUNT(b.id) 而非 COUNT(*)，否则 LEFT JOIN 未匹配的 NULL 行会被算成 1。
6. 线程安全验证：启动 5 个线程，各向 counter 表插入 100 行，验证最终行数是否为 500。如果出现少行，说明发生了什么？给出两种修复方案。
   参考答案要点：可能出现 sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread，或因共享连接导致的事务交错丢数据。修复：① 每线程一个连接（threading.local()）；② 用 check_same_thread=False 加一把全局 threading.Lock 保护所有操作。
7. SQLAlchemy ORM 实战：把 books / authors 改写成 ORM 模型，实现：创建作者及其图书、按作者名查询所有图书（并用 selectinload 避免 N+1）、统计每本书的借阅次数。
   参考答案要点：author = Author(name="余华"); author.books = [Book(title="活着"), Book(title="许三观卖血记")]; session.add(author); session.commit()。查全部书时用 session.scalars(select(Book).options(selectinload(Book.author))).all()；统计借阅次数用 select(func.count(Loan.id)).group_by(Loan.book_id)。
8. 实现带 TTL 的 Redis 缓存层：写一个 get_user_cached(uid)，命中 Redis 则直接返回，未命中则查 SQLite 并写入 Redis（TTL 300 秒）；再实现 update_user(uid, name) 更新数据库后删除对应缓存键。最后验证连续两次读取的耗时差异。
   参考答案要点：def get_user_cached(uid): key = f"user:{uid}"; c = r.get(key); if c: return json.loads(c); row = db.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone(); if row: r.setex(key, 300, json.dumps(dict(row))); return dict(row) if row else None。update 中先 db.commit() 再 r.delete(key)。用 time.perf_counter() 测量两次耗时。