Python 数据库事务与参数化查询、测试分层、迁移及发布检查。

Python 数据库、测试与发布

学习目标：把 Python API 从内存示例推进到持久化、可验证和可部署的服务。

1. 数据库访问先保证正确性

事务（transaction）把一组写操作作为一个整体提交或回滚。业务中的一次订单创建、库存扣减、支付记录等必须先明确哪些操作需要同成同败。

以下 SQLite 标准库示例可以直接运行；实际生产数据库的连接池、占位符和隔离行为应按驱动文档确认。

import sqlite3
from contextlib import closing

with closing(sqlite3.connect("notes.db")) as conn:
    with conn:
        conn.execute("CREATE TABLE IF NOT EXISTS note (id INTEGER PRIMARY KEY, title TEXT NOT NULL)")
        conn.execute("INSERT INTO note(title) VALUES (?)", ("第一条",))
    rows = conn.execute("SELECT id, title FROM note WHERE title = ?", ("第一条",)).fetchall()
print(rows)

? 占位符让参数与 SQL 语句分开，避免拼接用户输入。表结构变化要用迁移工具管理；在 Python 常见组合中，SQLAlchemy 2.x 负责 ORM/查询，Alembic 负责迁移。先理解 SQL 与事务，再使用 ORM。

易错点：不要把数据库异常直接返回给客户端；记录内部错误并返回稳定的错误格式。关闭连接和回滚也不能依赖“稍后由 GC 处理”。

2. 测试层次

层次	主要验证	示例
单元测试	纯业务规则	金额计算、状态转换
集成测试	数据库与仓储	约束、事务回滚、迁移
API 测试	路由与协议	状态码、输入校验、权限

def normalize_title(title: str) -> str:
    value = title.strip()
    if not value:
        raise ValueError("title is required")
    return value

def test_normalize_title():
    assert normalize_title("  hello  ") == "hello"

运行 pytest 时，测试数据库必须与真实数据隔离。每个测试可用独立数据库或事务回滚；覆盖重复键、并发写入、无权限和数据库暂时不可用。

3. 发布清单

固定依赖版本并确认 Python 版本；用环境变量配置数据库地址、密钥和日志级别。
启动前运行迁移，确认旧版本与新表结构是否兼容。
提供健康检查、结构化日志和请求 ID；区分“进程活着”与“依赖可用”。
明确服务器进程数、超时和优雅退出；不要在生产使用自动重载。
记录回滚步骤，发布后观察错误率和延迟，而不只看部署命令是否成功。

详见测试、CI 与部署和综合项目。

自测

参数化 SQL 能否替代权限检查？不能；它主要防止输入被当作 SQL 语法执行。
API 返回 201 是否证明数据真的写入？不能；还要验证持久化结果与失败时的回滚。

4. 一次写入的完整事务边界

以“创建笔记并记录审计事件”为例，两条写入要么都成功，要么都不生效。开启事务后先检查权限和版本，再执行 INSERT，最后写审计记录；任何一步失败都回滚。若审计写到外部消息系统，单个数据库事务无法跨系统保证原子性，需要考虑事务性 outbox 或失败重试。

import sqlite3
from contextlib import closing

def create_note(database: str, owner_id: int, title: str) -> int:
    with closing(sqlite3.connect(database)) as conn:
        with conn:
            cursor = conn.execute(
                "INSERT INTO note(owner_id, title) VALUES (?, ?)",
                (owner_id, title),
            )
            note_id = cursor.lastrowid
            if note_id is None:
                raise RuntimeError("未获得新笔记 ID")
            conn.execute(
                "INSERT INTO audit(note_id, action) VALUES (?, ?)",
                (note_id, "created"),
            )
            return note_id

示例假设两张表已由迁移创建。with conn: 管提交与回滚，closing 管连接关闭。生产服务应使用连接池或框架管理的会话，避免每次请求都随意建立新连接；SQLite 的写入并发行为也不能直接推断到 PostgreSQL 或 MySQL。

5. 测试失败路径而不污染环境

数据库测试应先运行迁移，再创建测试数据。至少验证：owner_id 不存在、标题违反约束、审计写入失败时笔记是否回滚、两个用户是否能越权读取。测试结果要可重复：测试数据库与开发数据库分离，测试彼此不依赖执行顺序，结束后清理临时资源。

使用 ORM 时也不能省略 SQL 理解。注意 Session 的生命周期、懒加载引发的 N+1 查询、更新丢失和唯一约束冲突。错误转换应区分“客户端可修正的冲突”与“内部数据库故障”，后者只写入受控日志，不泄漏 SQL、连接串或用户数据。

6. 从本机运行到可回滚发布

在干净环境创建虚拟环境并按锁定版本安装依赖；运行静态检查与测试。
用迁移工具升级独立数据库，确认新旧应用版本对表结构的兼容窗口。
部署同一构建产物，设置启动、就绪检查、请求超时和优雅关闭。
观察错误率、延迟与数据库连接数；异常时按预先写好的步骤回滚应用。数据库迁移若不可逆，应先设计向前修复方案。

03_数据库测试与发布