Python DB-API 与 sqlite3 · Python 数据库实战

第 8 章 · Python DB-API 2.0 与 sqlite3

本章目标：掌握 Python 标准库 sqlite3 与 PEP 249 DB-API 2.0 规范；理解 参数化查询 防 SQL 注入；使用 with 上下文管理器 管理连接与游标；在 db-demo 中对 shop_db 编写完整 CRUD 脚本；理解 slug + is_published + price（分） 在 Python 层的读写约定。

学时建议：4～5 小时（含 2 小时脚本跟练）

前置：完成 python-database ch07（shop_db DDL 已就绪）；python-dev ch10 文件 IO 与异常处理。

8.1 场景说明：不用 ORM 也要会访问数据库

在 db-demo 运维脚本、数据迁移小工具、单元测试夹具中，常需轻量直连 SQLite，而不引入 SQLAlchemy/Django。本章用标准库完成：

任务	脚本	技术
初始化库	`scripts/init_db.py`	执行 `schema/shop_db.sql`
插入分类/商品	`scripts/seed_products.py`	`INSERT` + 参数化
查询上架商品	`scripts/list_published.py`	`SELECT` + `Row`
更新库存	`scripts/update_stock.py`	`UPDATE` + 事务
删除下架草稿	`scripts/delete_draft.py`	`DELETE` + 确认

路径：~/python-learn/db-demo；数据库：data/shop_db.sqlite3。域名 api.example.com 仅为占位。

8.2 DB-API 2.0 核心概念

PEP 249 定义 Python 数据库接口标准，sqlite3、psycopg2、mysqlclient 均遵循：

对象	作用	sqlite3 示例
Connection	数据库连接	`sqlite3.connect(path)`
Cursor	执行 SQL、取结果	`conn.cursor()`
Transaction	提交/回滚	`conn.commit()` / `rollback()`
Parameter style	占位符	SQLite 用 `?` 或 `:name`

模块级属性：

import sqlite3

print(sqlite3.apilevel)    # '2.0'
print(sqlite3.threadsafety)  # 0～3，sqlite3 连接勿跨线程共享
print(sqlite3.paramstyle)    # 'qmark' → ?

`threadsafety`	含义
0	连接不可跨线程
1	连接可跨线程，游标不可
3	完全线程安全（少见）

8.3 连接数据库与 Row 工厂

scripts/db_config.py：

"""db-demo 共享数据库路径与连接工厂。"""
from __future__ import annotations

import sqlite3
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DB_PATH = ROOT / "data" / "shop_db.sqlite3"


def get_connection() -> sqlite3.Connection:
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row      # 行可 dict 式访问
    conn.execute("PRAGMA foreign_keys = ON")
    return conn

配置	说明
`row_factory = sqlite3.Row`	`row["slug"]`、`row.keys()`
`PRAGMA foreign_keys = ON`	启用 ch07 外键约束
`Path`	跨平台路径

最小连接示例：

from db_config import get_connection

with get_connection() as conn:
    cur = conn.execute("SELECT COUNT(*) AS n FROM categories")
    row = cur.fetchone()
    print(row["n"])

注意：sqlite3 3.12+ 的 connect() 支持 with 自动 commit/rollback；旧版本需手动 commit()。

8.4 上下文管理器：连接与游标

8.4.1 连接上下文

import sqlite3

with sqlite3.connect("shop.db") as conn:
    conn.execute("INSERT INTO categories (name, slug) VALUES (?, ?)", ("数码", "digital"))
    # 退出 with 块时：无异常 → commit；有异常 → rollback

8.4.2 显式游标上下文

with get_connection() as conn:
    with conn:                    # 事务边界
        cur = conn.cursor()
        try:
            cur.execute("UPDATE products SET stock = stock - ? WHERE id = ?", (2, 1))
            cur.execute("INSERT INTO order_lines (...) VALUES (...)", (...))
        finally:
            cur.close()

8.4.3 自定义上下文（可选）

from contextlib import contextmanager

@contextmanager
def shop_transaction():
    conn = get_connection()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

模式	适用
`with conn:`	单连接多语句事务
`@contextmanager`	脚本层统一 commit/rollback
无 with	仅临时查询、记得 `close()`

8.5 参数化查询与 SQL 注入防护

8.5.1 错误示范（禁止）

slug = "mouse'; DROP TABLE products; --"
sql = f"SELECT * FROM products WHERE slug = '{slug}'"  # 危险！
conn.execute(sql)

8.5.2 正确：占位符绑定

slug = "wireless-mouse"
cur = conn.execute(
    "SELECT id, title, price, is_published FROM products WHERE slug = ?",
    (slug,),
)
product = cur.fetchone()

命名参数（可读性更好）：

conn.execute(
    """
    INSERT INTO products (category_id, title, slug, price, stock, is_published)
    VALUES (:category_id, :title, :slug, :price, :stock, :is_published)
    """,
    {
        "category_id": 1,
        "title": "无线鼠标",
        "slug": "wireless-mouse",
        "price": 5990,           # 分
        "stock": 100,
        "is_published": 1,
    },
)

规则	说明
用户输入只通过 `?`/`:name` 传入	驱动负责转义
禁止 f-string 拼接 SQL 值	标识符（表名）若动态需白名单
`executemany`	批量插入

rows = [
    (1, "键盘", "mechanical-keyboard", 29900, 50, 1),
    (1, "鼠标垫", "mouse-pad-xl", 1990, 200, 1),
]
conn.executemany(
    """INSERT INTO products
       (category_id, title, slug, price, stock, is_published)
       VALUES (?, ?, ?, ?, ?, ?)""",
    rows,
)

8.6 价格（分）与 is_published 的 Python 约定

scripts/money.py：

def yuan_to_cents(yuan: str | float) -> int:
    """¥59.90 → 5990"""
    from decimal import Decimal, ROUND_HALF_UP
    d = Decimal(str(yuan))
    return int((d * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP))


def cents_to_yuan(cents: int) -> str:
    return f"¥{cents / 100:.2f}"


def as_bool_published(value: int | bool) -> bool:
    return bool(value)

场景	代码
展示	`cents_to_yuan(row["price"])`
写入	`yuan_to_cents("59.90")`
筛选上架	`WHERE is_published = 1`

8.7 CREATE / 初始化

scripts/init_db.py：

#!/usr/bin/env python3
"""从 schema/shop_db.sql 初始化 shop_db。"""
from pathlib import Path
from db_config import DB_PATH, get_connection

SCHEMA = Path(__file__).resolve().parents[1] / "schema" / "shop_db.sql"


def main() -> None:
    sql = SCHEMA.read_text(encoding="utf-8")
    if DB_PATH.exists():
        DB_PATH.unlink()
    with get_connection() as conn:
        conn.executescript(sql)
    print(f"Initialized {DB_PATH}")


if __name__ == "__main__":
    main()

executescript 一次执行多条 DDL；等价于 sqlite3 CLI 的 .read。