SQLite 导出只剩第一条：复用游标覆盖结果集的定位与修复

订单表有三条记录，导出任务正常结束，却只生成一条。没有 SQL 异常，也没有过滤条件变化。检查循环体后发现：遍历订单的游标，又被用来查询客户名称。

这类漏行先查游标的使用范围。增加超时或重试次数不会修复已经被替换的结果集。

适用场景与现象

本文针对 Python 标准库 sqlite3：外层逐条读取订单，内层查询关联信息，两个查询误用了同一个 Cursor。示例只读数据库，不涉及 ORM、线程共享连接或远程数据库驱动。

典型信号是：源订单数量大于导出行数，任务没有异常；内层用 fetchone() 读完唯一客户后，外层恰好结束。若内层查询返回多行而没有读完，外层甚至可能开始读取客户查询剩余的行，随后出现字段数量不符。

原因：结果属于游标当前执行的查询

Python 的 SQLite 游标既执行 SQL，也维护取数上下文；它可以直接作为迭代器。for row in cursor.execute(...) 不会提前复制所有行。循环体再次对同一个游标执行 SQL，就不能继续把它当作原订单结果集来读。接口定义见 Python sqlite3 游标文档。

排查时沿着外层循环检查所有辅助函数，特别留意把 cursor 作为参数传入的“查询名称”“读取配置”函数。变量名不同也不代表对象不同：lookup = cursor 仍然是同一个游标。

诊断时可记录 source_count、exported_count、query_count 等数字和固定中文消息，不必输出客户名称或完整订单。不要用 SELECT 的 rowcount 作为总行数；计数应使用相同条件的 COUNT(*)，或在受控数据集里比较主键集合。

可直接运行的复现与修复

保存为 cursor_export_demo.py，执行 python cursor_export_demo.py。仅使用标准库，在内存数据库内完成，不会连接生产环境。

示例把订单的 customer_id 设为可空，保留未关联客户的订单；客户表主键保证每笔订单至多匹配一个客户。这两个条件决定后面 JOIN 的正确语义。

import sqlite3
import unittest
from contextlib import closing


def create_database() -> sqlite3.Connection:
    """创建独立测试数据，保留无客户关联的订单以验证导出完整性。"""
    connection = sqlite3.connect(":memory:")
    connection.executescript(
        """
        CREATE TABLE customer (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL
        );
        CREATE TABLE sales_order (
            id INTEGER PRIMARY KEY,
            customer_id INTEGER
        );
        INSERT INTO customer VALUES (1, '客户甲'), (2, '客户乙');
        INSERT INTO sales_order VALUES (101, 1), (102, 2), (103, NULL);
        """
    )
    return connection


def export_broken(connection: sqlite3.Connection) -> list[tuple[int, str]]:
    """保留故障写法，用于证明无异常结束也可能漏行。"""
    result = []
    with closing(connection.cursor()) as cursor:
        for order_id, customer_id in cursor.execute(
            "SELECT id, customer_id FROM sales_order ORDER BY id"
        ):
            customer = cursor.execute(
                "SELECT name FROM customer WHERE id = ?", (customer_id,)
            ).fetchone()
            result.append((order_id, customer[0] if customer else '未关联'))
    return result


def export_separate(connection: sqlite3.Connection) -> list[tuple[int, str]]:
    """拆分取数上下文，适合先修复既有循环中的游标复用问题。"""
    result = []
    with closing(connection.cursor()) as orders, closing(
        connection.cursor()
    ) as customers:
        for order_id, customer_id in orders.execute(
            "SELECT id, customer_id FROM sales_order ORDER BY id"
        ):
            customer = customers.execute(
                "SELECT name FROM customer WHERE id = ?", (customer_id,)
            ).fetchone()
            result.append((order_id, customer[0] if customer else '未关联'))
    return result


def export_join(connection: sqlite3.Connection) -> list[tuple[int, str]]:
    """通过多对一关联一次读取，避免逐订单追加客户查询。"""
    with closing(connection.cursor()) as cursor:
        return cursor.execute(
            """
            SELECT o.id, COALESCE(c.name, '未关联')
            FROM sales_order AS o
            LEFT JOIN customer AS c ON c.id = o.customer_id
            ORDER BY o.id
            """
        ).fetchall()


class CursorExportTests(unittest.TestCase):
    """同时检查主键和关联内容，防止只比较行数掩盖错误。"""

    def setUp(self) -> None:
        self.connection = create_database()
        self.addCleanup(self.connection.close)
        self.expected = [(101, '客户甲'), (102, '客户乙'), (103, '未关联')]

    def test_reproduce_missing_rows(self) -> None:
        self.assertEqual(export_broken(self.connection), self.expected[:1])

    def test_separate_cursors(self) -> None:
        self.assertEqual(export_separate(self.connection), self.expected)

    def test_join(self) -> None:
        self.assertEqual(export_join(self.connection), self.expected)

    def test_empty_orders(self) -> None:
        self.connection.execute("DELETE FROM sales_order")
        for exporter in (export_separate, export_join):
            with self.subTest(exporter=exporter.__name__):
                self.assertEqual(exporter(self.connection), [])

    def test_missing_customer(self) -> None:
        self.connection.execute("DELETE FROM customer WHERE id = ?", (1,))
        expected = [(101, '未关联'), (102, '客户乙'), (103, '未关联')]
        for exporter in (export_separate, export_join):
            with self.subTest(exporter=exporter.__name__):
                self.assertEqual(exporter(self.connection), expected)

    def test_multiple_orders_same_customer(self) -> None:
        self.connection.execute("INSERT INTO sales_order VALUES (?, ?)", (104, 1))
        expected = self.expected + [(104, '客户甲')]
        self.assertEqual(export_separate(self.connection), expected)
        self.assertEqual(export_join(self.connection), expected)

    def test_database_error_propagates(self) -> None:
        self.connection.execute("DROP TABLE customer")
        for exporter in (export_separate, export_join):
            with self.subTest(exporter=exporter.__name__):
                with self.assertRaises(sqlite3.OperationalError):
                    exporter(self.connection)

    def test_query_count(self) -> None:
        statements: list[str] = []
        self.connection.set_trace_callback(statements.append)
        try:
            export_separate(self.connection)
            self.assertEqual(len(statements), 4)
            statements.clear()
            export_join(self.connection)
            self.assertEqual(len(statements), 1)
        finally:
            self.connection.set_trace_callback(None)


if __name__ == "__main__":
    unittest.main(verbosity=2)

第一项测试故意断言错误实现只返回第一单，作为稳定的故障复现；修复正确性由后续测试验证。正常预期共八项测试通过。test_query_count 验证三笔订单的独立游标版本执行四次查询，JOIN 版本执行一次；这个计数是本地语句数量，不是性能压测结果。跟踪回调仅用于合成测试数据，不应直接用于记录生产 SQL 参数。

两种修复如何选择

已有业务循环很复杂时，先拆分游标容易做小范围修复。若客户名称只是一个数据库字段，优先考虑 JOIN，但必须验证订单主键集合与关联内容，不能只看总行数一致。

这里采用 LEFT JOIN 是为了保留无关联订单；如果改成 INNER JOIN，测试里的第 103 单会消失。不能把 JOIN 生搬到一对多明细表：那会产生多行订单，需要另外设计聚合或导出粒度。

另一种临时办法是先对订单 fetchall() 再循环，但它把全部订单装进内存，仅适合数量已有明确上限的小批次。本示例返回列表是为便于断言；百万行导出应在单个查询游标上迭代或用 fetchmany() 分批写文件，并让该游标保持专用。

上线前验证与失败处理

1. 用固定数据覆盖零行、多行、客户缺失和多个订单共享客户，断言订单 ID 的集合、顺序以及名称；不能只验证任务退出码。
2. 导出期间需要跨多次查询的一致视图时，由调用方明确事务边界。独立游标只分离取数状态，不会自动解决并发更新的一致性问题。
3. 不要在读取结果尚未结束时，用同一连接修改被扫描的数据并假设行为不变。SQLite 对这种可见性有专门限制，见 SQLite 隔离说明。本文的两个修复函数均只读。
4. 查询或文件写入失败应让任务失败，不应捕获异常后把部分结果标成成功。文件导出可先写临时文件，关闭并验证后再替换正式文件，避免用户下载到半份报告。
5. 发布时保留旧程序包用于回滚，但已漏行的报告仍需重新生成。先确定受影响时间、过滤条件和订单集合，再重跑；换回旧代码不会补回历史文件。

本次验证范围

正文代码在 Python 3.14.6、SQLite 3.50.4 环境执行，八项 unittest 验证游标覆盖复现、两种修复、空集、缺少关联、共享客户、数据库异常传播和查询数量。它没有验证线上订单、文件系统原子替换、并发快照、ORM 或其他数据库驱动；这些需要在实际接入时另外验收。

游标复用的危险在于任务可能正常结束。为外层结果保留专用游标，并以业务主键集合验证输出完整性，才能把这种静默漏行从“任务成功”里识别出来。