数据库编程学习路线：连接数据库、增删改查与事务

数据库编程不只是记住几条 SQL。实际开发中，程序还要管理连接、传递参数、处理并发修改、保证多步操作的一致性，并在失败时妥善回滚。学习时如果只练习“查出一行数据”，很容易在正式项目里遇到 SQL 注入、事务未提交、连接未关闭等问题。

本文按从易到难的顺序介绍数据库编程：先用 SQLite 建立可运行的练习环境，再掌握连接与 CRUD（增删改查），随后学习事务、异常处理和常见边界。示例使用 Python 标准库，无须安装额外数据库驱动，适合入门和小型本地程序。理解这些概念后，再迁移到 PostgreSQL、MySQL 等服务端数据库会更容易。

一、先认识数据库、SQL 与驱动

数据库负责持久化数据，并提供查询、约束和并发访问等能力。SQL 是与关系型数据库交互的语言；数据库驱动则是应用程序与某一种数据库通信的接口。应用通常通过驱动建立连接，创建游标或执行器，发送 SQL 和参数，然后读取结果。

学习路线可以分为四步：

1. 理解数据模型：表、列、行、主键、唯一约束、外键和索引。
2. 掌握 SQL：SELECT、INSERT、UPDATE、DELETE，以及 WHERE、ORDER BY、JOIN、GROUP BY。
3. 从程序调用 SQL：建立连接、参数化执行、读取结果、处理错误、关闭资源。
4. 学习可靠性与性能：事务、并发、索引、连接池、迁移和备份。

SQLite 把数据库存放在文件中，适合原型、桌面应用、测试和单机工具。它不要求另行启动数据库服务，因而适合作为练习入口。多用户服务、需要集中管理权限或高并发写入的应用，通常应评估 PostgreSQL、MySQL 等服务端数据库。SQLite 和服务端数据库在类型、并发控制、SQL 方言和运维方式上并不完全相同；用 SQLite 学会的程序结构可以迁移，具体 SQL 和部署假设则不能一概照搬。

二、准备可运行的练习环境

安装 Python 3 后，将下面代码保存为 db_lesson.py，运行 python db_lesson.py。程序会在当前目录创建 lesson.db。SQLite 数据库文件可以在程序退出后保留，因此重复运行时要考虑已有数据。后文会逐步替换或扩充这份示例。

import sqlite3

DB_PATH = "lesson.db"

def connect():
    # with 管理事务提交或回滚；连接仍需显式关闭。
    return sqlite3.connect(DB_PATH)

def initialize():
    conn = connect()
    try:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY,
                title TEXT NOT NULL,
                done INTEGER NOT NULL DEFAULT 0
                    CHECK (done IN (0, 1))
            )
        """)
        conn.commit()
    finally:
        conn.close()

if __name__ == "__main__":
    initialize()
    print("数据库已初始化：", DB_PATH)

CREATE TABLE IF NOT EXISTS 便于重复运行初始化代码，但它不是完整的数据库迁移工具：如果表已经存在而结构发生变化，这条语句不会自动调整列或约束。项目结构变化时，应采用明确的迁移脚本或迁移工具，并在部署时按顺序应用变更。

这个表包含三列：id 是主键，title 不允许为空，done 用整数表示未完成或已完成，并由 CHECK 限制只能取 0 或 1。主键和约束应尽量放在数据库中定义，因为数据库是最终写入数据的地方，不能只依赖页面或应用代码校验。

三、连接、游标与资源管理

程序中的“连接”代表与数据库的一段会话。在 SQLite 中，连接对象可以直接执行 SQL，适合简单示例；很多数据库驱动还会提供游标对象，用它执行语句并逐行读取结果。服务端数据库的连接通常要通过连接池复用，而不是为每一次查询都新建一条网络连接。

连接需要及时释放。Python 的 SQLite 连接可用 close() 关闭。with conn: 主要用于管理事务：离开代码块时，成功则提交，发生异常则回滚；它本身并不保证关闭连接。因此，示例使用 try/finally 显式关闭连接。也可以写一个上下文管理器，同时管理连接的开启与关闭。

在长时间运行的服务中，不要把一个连接随意跨线程或跨请求共享。连接是否可跨线程、能否并行执行操作，取决于驱动和数据库配置；连接池也要遵照所用框架或驱动的线程安全约定。新手练习时，一个操作函数创建连接、完成工作并关闭连接，通常更容易理解；生产环境再按实际负载引入连接池。

四、CRUD：插入、查询、更新与删除

1. 插入数据：使用占位符绑定参数

SQL 语句中的占位符应与数据参数分开传入。Python 的 sqlite3 使用问号占位符；参数以序列传递。单个参数也要写成单元素元组，注意末尾逗号。不要用字符串拼接把用户输入嵌进 SQL。

import sqlite3

DB_PATH = "lesson.db"

def add_task(title):
    conn = sqlite3.connect(DB_PATH)
    try:
        cursor = conn.execute(
            "INSERT INTO tasks (title) VALUES (?)",
            (title,)
        )
        conn.commit()
        return cursor.lastrowid
    finally:
        conn.close()

if __name__ == "__main__":
    task_id = add_task("学习参数化查询")
    print("新增任务 ID：", task_id)

参数绑定的好处不只是转义特殊字符。驱动会把参数作为数据传递，而不是让它们变成 SQL 语法的一部分，从而避免常见的 SQL 注入风险，也能正确处理引号等字符。仍然需要对业务数据做验证，例如限制标题长度、拒绝空白标题；参数化不能代替业务校验、授权检查或数据约束。

占位符通常只能绑定值，不能用于表名、列名、SQL 关键字或排序方向。例如下面写法不会把表名参数化：

conn.execute("SELECT * FROM ?", (table_name,))

如果确实要动态选择列或排序方向，应将输入限制在程序内预先定义的白名单中，再据此构造 SQL 语句。不要直接拼接未经验证的用户输入。不同数据库驱动的占位符格式可能不同，例如有的使用 %s 或命名占位符；应查阅当前驱动文档，不能把 SQLite 的问号形式原样复制到所有驱动。

2. 查询数据：fetchone、fetchall 与结果列

查询使用 SELECT。fetchone() 取一行，没有结果时返回 None；fetchall() 取回剩余所有行，返回列表。数据量较大时，直接 fetchall() 会把结果全部放进内存，应考虑分页或逐行迭代。

import sqlite3

DB_PATH = "lesson.db"

def get_task(task_id):
    conn = sqlite3.connect(DB_PATH)
    try:
        row = conn.execute(
            "SELECT id, title, done FROM tasks WHERE id = ?",
            (task_id,)
        ).fetchone()

        if row is None:
            return None

        return {
            "id": row[0],
            "title": row[1],
            "done": bool(row[2]),
        }
    finally:
        conn.close()

if __name__ == "__main__":
    print(get_task(1))

显式列出所需列，比 SELECT * 更清楚，也能减少表结构变化带来的意外影响。若想按列名访问结果，可以为连接设置行工厂：

import sqlite3

conn = sqlite3.connect("lesson.db")
conn.row_factory = sqlite3.Row
try:
    row = conn.execute(
        "SELECT id, title, done FROM tasks WHERE id = ?",
        (1,)
    ).fetchone()
    if row is not None:
        print(row["title"], row["done"])
finally:
    conn.close()

查询多行时，应明确排序条件。关系表中的行没有天然顺序；没有 ORDER BY 时，不应假定结果总以某种固定顺序返回。

3. 更新数据：限制目标行并检查影响行数

更新语句通常需要 WHERE。漏写条件可能会修改整张表。下面的函数只更新指定任务，并通过 rowcount 判断是否有行受到影响：

import sqlite3

DB_PATH = "lesson.db"

def mark_done(task_id, done=True):
    conn = sqlite3.connect(DB_PATH)
    try:
        cursor = conn.execute(
            "UPDATE tasks SET done = ? WHERE id = ?",
            (1 if done else 0, task_id)
        )
        conn.commit()
        return cursor.rowcount
    finally:
        conn.close()

if __name__ == "__main__":
    changed = mark_done(1)
    print("受影响行数：", changed)

rowcount 的具体含义可能随数据库和驱动而异，尤其是更新值与旧值相同的情况。它常用于检查是否匹配到预期记录，但不能在所有数据库中都简单理解为“值实际发生变化的行数”。业务需要严格判断时，要根据驱动文档和数据库行为验证，必要时使用返回子句或再查询确认。

4. 删除数据：明确范围并关注外键

删除也必须谨慎使用 WHERE。数据库约束、级联删除规则会影响删除行为；若开启外键约束，删除被引用的记录可能失败，也可能按已定义的级联规则影响关联表。

import sqlite3

DB_PATH = "lesson.db"

def delete_task(task_id):
    conn = sqlite3.connect(DB_PATH)
    try:
        cursor = conn.execute(
            "DELETE FROM tasks WHERE id = ?",
            (task_id,)
        )
        conn.commit()
        return cursor.rowcount
    finally:
        conn.close()

if __name__ == "__main__":
    print("删除行数：", delete_task(1))

在业务系统中，删除前应确认权限、关联数据和审计需求。软删除（增加状态列、保留记录）与物理删除适用于不同场景，不能只为“方便恢复”就默认采用软删除；软删除会增加查询过滤和唯一约束设计的复杂度。

五、输入验证、分页与动态 SQL 的边界

绑定参数解决的是“数据被当成 SQL 代码执行”的问题，不会自动限制输入长度、格式、业务范围或访问权限。比如用户只能修改自己的任务，应用必须在查询或更新条件中加入用户身份约束，并在服务端确认当前用户有权操作；不能仅凭客户端传来的任务 ID 就认定授权成立。

分页查询可以限制一次返回的数据量。下面展示 SQLite 的常见写法：

import sqlite3

DB_PATH = "lesson.db"

def list_tasks(limit, offset):
    # limit 和 offset 由程序校验，拒绝负数并设置合理上限。
    if not isinstance(limit, int) or not 1 <= limit <= 100:
        raise ValueError("limit 必须是 1 到 100 之间的整数")
    if not isinstance(offset, int) or offset < 0:
        raise ValueError("offset 必须是非负整数")

    conn = sqlite3.connect(DB_PATH)
    try:
        return conn.execute(
            "SELECT id, title, done FROM tasks "
            "ORDER BY id LIMIT ? OFFSET ?",
            (limit, offset)
        ).fetchall()
    finally:
        conn.close()

if __name__ == "__main__":
    for row in list_tasks(20, 0):
        print(row)

这里的上下限是示例程序设定的策略，不是 SQLite 对 LIMIT 参数的通用最大值。限制分页大小是为了控制应用一次处理的数据量。不同数据库对分页语法、参数类型和效率的处理可能不同；大偏移量分页可能越来越慢，数据量很大时可评估基于排序键的游标式分页。

不要因为某个参数在当前数据库中可以绑定，就省略校验。对于动态排序字段，使用白名单映射，例如将用户可选的“标题”映射到固定 SQL 列名，将“时间”映射到另一列名；排序方向只允许预设的 ASC 或 DESC。值继续通过占位符传递，标识符则通过经过验证的固定选项决定。

六、事务：把多个操作作为一个整体

事务用于把一组数据库操作作为一个整体处理。常用的概括是 ACID：原子性（要么全部生效，要么全部回滚）、一致性（遵守约束与业务不变量）、隔离性（并发事务互相影响的程度由数据库隔离机制决定）、持久性（提交后的数据按数据库保证保存）。实际隔离级别和细节依数据库而异，不能仅凭 ACID 这四个词推断所有并发行为。

例如创建订单时，可能既要插入订单，也要扣减库存。如果订单写入成功而库存更新失败，数据就会不一致。将两步放在同一事务中，任一步失败就回滚。下面的示例使用两张表，并在一个事务内完成插入订单与扣减库存。

import sqlite3

DB_PATH = "lesson.db"

def initialize_order_tables():
    conn = sqlite3.connect(DB_PATH)
    try:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS inventory (
                sku TEXT PRIMARY KEY,
                stock INTEGER NOT NULL CHECK (stock >= 0)
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS orders (
                id INTEGER PRIMARY KEY,
                sku TEXT NOT NULL,
                quantity INTEGER NOT NULL CHECK (quantity > 0)
            )
        """)
        conn.commit()
    finally:
        conn.close()

def place_order(sku, quantity):
    if not isinstance(quantity, int) or quantity <= 0:
        raise ValueError("quantity 必须是正整数")

    conn = sqlite3.connect(DB_PATH)
    try:
        # 上下文块中的异常会触发回滚，正常结束则提交。
        with conn:
            cursor = conn.execute(
                "UPDATE inventory "
                "SET stock = stock - ? "
                "WHERE sku = ? AND stock >= ?",
                (quantity, sku, quantity)
            )

            if cursor.rowcount != 1:
                raise ValueError("商品不存在或库存不足")

            order_cursor = conn.execute(
                "INSERT INTO orders (sku, quantity) VALUES (?, ?)",
                (sku, quantity)
            )
            return order_cursor.lastrowid
    finally:
        conn.close()

if __name__ == "__main__":
    initialize_order_tables()
    print("请先向 inventory 表添加商品后再创建订单。")

示例把“库存是否足够”写进更新条件，而不是先读取库存、再在程序中判断、最后更新。这样可以减少检查与更新之间的竞态窗口；数据库的写入锁、隔离机制及约束仍然是整体设计的一部分。库存扣减成功但订单插入失败时，事务回滚也会撤销扣减。

示例最后的 print 只是提醒准备数据，不会自动创建商品。需要先执行下面的初始化数据语句，才能成功调用 place_order：

import sqlite3

conn = sqlite3.connect("lesson.db")
try:
    with conn:
        conn.execute(
            "INSERT OR IGNORE INTO inventory (sku, stock) VALUES (?, ?)",
            ("book-001", 10)
        )
finally:
    conn.close()

# 准备好库存后运行：
# print(place_order("book-001", 2))

INSERT OR IGNORE 是 SQLite 的冲突处理语法，重复的主键会被忽略；它不是所有数据库都支持的通用 SQL。迁移到其他数据库时，应使用该数据库对应的冲突处理语法，并确认忽略冲突是否符合业务预期。

事务不应包住不必要的长时间工作。不要在持有数据库事务期间等待用户输入、下载文件或调用慢速外部服务，否则会延长锁的占用时间。数据库事务也不能自动撤销已经发出的电子邮件或 HTTP 请求。涉及外部系统的业务通常需要设计补偿、重试或可靠消息机制，而不是假设数据库回滚可以回滚世界上的所有操作。

七、异常处理：回滚、重试与错误分类

数据库操作可能因为约束冲突、连接中断、SQL 错误或并发竞争而失败。异常应在合适的应用边界处理：可以记录必要的诊断信息并返回明确错误，但不要把密码、完整连接串、敏感用户数据或内部 SQL 细节直接展示给最终用户。

事务代码中不应悄悄吞掉异常。示例让异常从 with conn: 代码块中传出，因此事务管理器可以执行回滚，调用者也知道操作失败。捕获异常后如果要继续处理，应明确区分可恢复错误和不可恢复错误；不能捕获所有异常、打印一句话，然后当作成功返回。

重试不是通用的错误处理方式。连接超时、死锁等错误在特定数据库环境下可能适合有限次数重试，但需要判断事务是否已经提交，以及操作是否幂等。盲目重试一个“扣款”操作可能造成重复执行。每个数据库及驱动对错误码、断线后的状态和重试建议都有自己的说明，应以其文档为准。

八、连接和事务中的常见错误

• 直接拼接用户输入：容易产生 SQL 注入，也会因引号、编码等问题导致语句失败。数据值使用参数绑定；动态标识符采用白名单。
• 忘记提交：某些驱动默认使用事务，写入后没有提交，连接关闭时可能回滚。应了解驱动的事务默认行为，并明确事务边界。
• 只管理事务，不关闭连接：with conn: 在 SQLite 中管理事务，但连接仍须关闭。使用 finally 或可靠的上下文管理器释放资源。
• UPDATE 或 DELETE 漏掉 WHERE：可能影响全表。上线前审查 SQL；高风险操作应增加保护条件、审核流程或备份策略。
• 把 SQLite 当成所有数据库的缩小版：驱动 API、SQL 方言、类型、锁和并发能力存在差异。换数据库时应重新检查事务、分页和冲突处理语法。
• 只在应用层校验：并发写入和其他程序也可能违反规则。对可由数据库表达的规则，使用 NOT NULL、UNIQUE、CHECK、外键等约束，并保留应用层友好提示。
• 查询结果没有稳定排序：没有 ORDER BY，不能依赖记录返回顺序。分页和展示应指定明确的排序键。
• 把连接信息硬编码进代码：真实项目应通过环境变量或受控的秘密管理机制提供凭据，并限制数据库账号权限。SQLite 本地文件还需要考虑文件系统权限与备份。

九、从练习代码走向项目代码

从示例迈向真实项目时，可以按下面的顺序扩展，不必一开始就引入复杂架构：

1. 拆分职责：把连接配置、数据库初始化和业务查询分开，避免所有逻辑堆在一个脚本中。
2. 设计约束与索引：根据查询和数据关系设置主键、外键、唯一约束及必要索引。索引可以加速部分查询，但会占用空间并增加写入维护成本，不应盲目给每列建索引。
3. 加入迁移：用可追踪、可重复部署的变更脚本更新表结构，避免只依赖首次建表代码。
4. 测试失败路径：验证空输入、重复记录、找不到目标行、约束冲突和事务回滚。数据库测试最好使用隔离的临时库或独立测试数据库，避免误改开发数据。
5. 评估并发和连接池：服务端应用应使用驱动或框架提供的连接池，设置合理的连接数和超时，并根据实际数据库监控调整。
6. 监控与备份：关注慢查询、连接失败、锁等待、磁盘空间和备份恢复流程。备份文件存在并不等于能恢复，应定期验证恢复步骤。

学习 ORM（对象关系映射）可以提高日常开发效率，但 ORM 不会消除 SQL、事务和索引知识的重要性。理解 ORM 实际生成的 SQL，尤其是关联查询、批量写入和事务边界，能够避免隐蔽的性能问题。无论使用 ORM 还是直接编写 SQL，参数绑定、授权、约束和连接管理都不可省略。

十、用一个小项目巩固

可以把本文的任务表扩展成一个命令行待办程序：新增任务、列出任务、按 ID 标记完成、删除任务，再加入用户或分类。实现时逐项检查：

• 查询是否显式选择列，并为分页和列表指定稳定排序？
• 写操作是否使用参数绑定，更新和删除是否有明确范围？
• 业务规则是否同时考虑应用验证与数据库约束？
• 一组必须同时成功的写操作是否放入同一事务？
• 连接是否在成功和失败路径都能关闭？
• 测试是否覆盖重复数据、无效输入和事务回滚？

数据库编程的核心不是背诵 API，而是把程序状态与数据状态可靠地衔接起来。先用 SQLite 练习连接、参数化 CRUD 和事务，再根据项目需要学习目标数据库的驱动、隔离级别、锁、迁移及运维方式。每次写数据库代码时，明确“参数是什么、修改范围是什么、失败时如何回滚、谁有权操作”，就能避开多数入门阶段的严重问题。