用 SQLite 复合游标做稳定的记录列表翻页

记录列表已经能按发布时间倒序显示。每页两条，首屏是 106、105。用户停了一会儿，这期间又发布了一条新记录；点击“下一页”，105 却出现了第二次。下面把这个现象做成可以运行的小实验，再实现一个按排序边界继续读取的分页函数。

本例的结果是一份 Python 标准库 + SQLite 记录列表实现，返回当前页、下一页游标和 has_more。我们会让三条记录落在同一秒，分别测试插入、删除和修改排序字段。它适合“加载更多”或顺序浏览；需要直接跳到第 200 页、固定总页数或严格一致的导出时，要另定协议。

需要 Python 3.12 及以上，因为连接明确使用 autocommit=True；行值比较需要 SQLite 3.15.0 及以上。Python 自带的 sqlite3 模块所链接的 SQLite 版本要单独检查，不能从 Python 版本推断。本次在 Python 3.12.14 / SQLite 3.53.1 上实际运行了后面的 22 项测试，全部使用临时数据库和合成记录。

python -c "import sys, sqlite3; print(sys.version); print(sqlite3.sqlite_version)"

文中时间都属于合成数据，统一为 2026-10-12 UTC。published 存整数 Unix 秒，允许范围为 0 到 SQLite 有符号 64 位整数上限；这一简化不包含 1970 年之前的日期。资料核对日期为 2026-10-13。

列表固定按 published DESC, id DESC 排序。published 表示发布时间，id 是唯一正整数。先比较时间；时间相同，再比较 id。SQLite 明确说明：若所有 ORDER BY 表达式都相等，行的相对顺序没有保证。把唯一 id 放在最后，才能为这份列表定义完整顺序。SQLite：ORDER BY 规则

游标保存上一页最后一条“实际展示记录”的 published 和 id，同时携带 feed，避免在不同列表间误用。本例限定两列都是倒序，因此向后读取时使用严格小于：

SELECT id, published, title
FROM records
WHERE feed = ? AND (published, id) < (?, ?)
ORDER BY published DESC, id DESC
LIMIT ?;

这个比较从左到右进行：发布时间更早的记录可以进入；时间相同的记录只有 id 更小时才进入。SQLite 官方的滚动窗口例子采用相同的多列边界思路，本文将它具体化为双列倒序的发布列表。SQLite：Scrolling Window Queries

不能顺手把两列改成一升一降后仍用同一个 < 条件。混合方向、可空字段、自定义排序规则或其他过滤条件，都需要重新推导边界并补测试。这里让 published 非空且为整数，先排除日期字符串格式和 NULL 比较带来的歧义。

主列表只有六条记录，顺序为 106、105、104、103、102、101。其中 105、104、103 都在 10:01:00 发布。另放一条属于 other 列表的 999，确认过滤条件没有在第二页消失。每个实验都重新建库，避免上一个实验的改动影响下一个。

先取两条，再在顶部插入 107。原来的第二页查询仍是跳过两条，但它跳过的是当前列表里的 107、106，于是返回 105、104。保存的复合游标仍指向原来的 (10:01:00, 105)，接下去就得到 104、103。

-- 负例：它描述当前结果中的位置，不保存上一次阅读边界。
SELECT id, published, title FROM records
WHERE feed = ?
ORDER BY published DESC, id DESC
LIMIT ? OFFSET ?;
-- 绑定 ("main", 2, 2)：插入 107 后，得到 105、104。

这个实验只证明此类插入对翻页正确性的影响。它没有做查询基准，也不推出“某个页数以后一定慢”的阈值。示例索引按 feed、published、id 排列，匹配固定过滤与排序；真实项目仍需根据数据量、其他条件和实际执行计划验证。

把下面文件保存为 pagination.py。动态的 feed、游标值和 LIMIT 全部通过 ? 占位符绑定；排序列和方向固定写在 SQL 中。Python 官方文档建议用参数替换绑定值，避免拼接输入。占位符不能代替列名或 ASC / DESC 等 SQL 结构。Python：绑定查询参数

页大小为 2 时，最多读取 3 行。前 2 行返回给界面，第 3 行只判断本次查询是否还有后续数据。下一页游标取第 2 行，不能取第 3 行，否则会把探测到的记录跳过去。没有额外行时，has_more 为 false，next_cursor 为 None。

"""向后翻页：固定 published DESC, id DESC，所有动态值均绑定参数。
Python 3.12+；仅用标准库。Cursor 是应用内部值对象，不是鉴权凭证。
"""
from dataclasses import dataclass
import sqlite3


@dataclass(frozen=True)
class Record:
    id: int
    published: int
    title: str


@dataclass(frozen=True)
class Cursor:
    feed: str
    published: int
    id: int


@dataclass(frozen=True)
class Page:
    items: tuple[Record, ...]
    next_cursor: Cursor | None
    has_more: bool


def connect(path: str) -> sqlite3.Connection:
    # 使每条完成的 SELECT 独立结束；本例不会跨分页请求保留事务。
    if sqlite3.sqlite_version_info < (3, 15, 0):
        raise RuntimeError("row-value comparison requires SQLite 3.15.0+")
    return sqlite3.connect(path, autocommit=True)


def _integer(value: object, name: str, minimum: int) -> None:
    # bool 是 int 的子类，但不接受 True 作为页大小或时间。
    if type(value) is not int or not minimum <= value <= 2**63 - 1:
        raise ValueError(f"invalid {name}")


def list_page(
    conn: sqlite3.Connection,
    feed: str,
    cursor: Cursor | None = None,
    limit: int = 2,
) -> Page:
    if not isinstance(feed, str) or not 1 <= len(feed) <= 128:
        raise ValueError("invalid feed")
    if type(limit) is not int or not 1 <= limit <= 100:
        raise ValueError("limit must be an integer in [1, 100]")

    if cursor is None:
        sql = """
            SELECT id, published, title FROM records
            WHERE feed = ?
            ORDER BY published DESC, id DESC
            LIMIT ?
        """
        params = (feed, limit + 1)
    else:
        if not isinstance(cursor, Cursor) or cursor.feed != feed:
            raise ValueError("cursor belongs to another feed or is invalid")
        _integer(cursor.published, "cursor.published", 0)
        _integer(cursor.id, "cursor.id", 1)
        sql = """
            SELECT id, published, title FROM records
            WHERE feed = ? AND (published, id) < (?, ?)
            ORDER BY published DESC, id DESC
            LIMIT ?
        """
        params = (feed, cursor.published, cursor.id, limit + 1)

    # 多取一行只探测本次查询还有没有数据，游标取展示行的最后一行。
    rows = conn.execute(sql, params).fetchall()
    items = tuple(Record(*row) for row in rows[:limit])
    has_more = len(rows) > limit
    next_cursor = None
    if has_more:
        last = items[-1]
        next_cursor = Cursor(feed, last.published, last.id)
    return Page(items, next_cursor, has_more)

Cursor 是我们定义的分页值对象，与 sqlite3.Cursor 查询对象不是一回事。这里只在 Python 函数之间传值，没有实现 URL token、签名或权限检查。输入类型、整数范围和列表作用域检查有助于尽早报错，不能替代服务端鉴权。

把下面内容保存为同目录的 fixtures.py。表结构禁止空发布时间，整数存储检查让这个小例子的比较规则保持一致。id 必须唯一；实际列表中也应让 id 保持不变，并明确发布时间能否编辑。

database() 每次创建独立临时目录，关闭连接后清理文件。运行示例不需要你提供任何数据库路径，也不访问网络。请先在这组数据上验收，不要把实验中的插入、更新和删除改到线上库执行。

"""纯合成数据；所有时间是 UTC Unix 秒，不关联真实用户或线上数据库。"""
from datetime import datetime, timezone
from contextlib import contextmanager
from pathlib import Path
from tempfile import TemporaryDirectory
from pagination import connect


def stamp(hour: int, minute: int, second: int = 0) -> int:
    return int(datetime(2026, 10, 12, hour, minute, second,
                        tzinfo=timezone.utc).timestamp())


SCHEMA = """
CREATE TABLE records (
    id INTEGER PRIMARY KEY CHECK (id > 0),
    feed TEXT NOT NULL,
    published INTEGER NOT NULL
        CHECK (typeof(published) = 'integer' AND published >= 0),
    title TEXT NOT NULL
);
CREATE INDEX records_feed_published_id
    ON records(feed, published DESC, id DESC);
"""
ROWS = [
    (106, "main", stamp(10, 2), "版本发布"),
    (105, "main", stamp(10, 1), "文档更新"),
    (104, "main", stamp(10, 1), "示例补充"),
    (103, "main", stamp(10, 1), "链接修正"),
    (102, "main", stamp(10, 0), "测试通过"),
    (101, "main", stamp(9, 59), "初始记录"),
    (999, "other", stamp(10, 9), "其他列表，不应混入"),
]


def seed(conn):
    conn.executescript(SCHEMA)
    conn.executemany(
        "INSERT INTO records(id, feed, published, title) VALUES (?, ?, ?, ?)",
        ROWS,
    )


@contextmanager
def database():
    # 每个场景独占临时目录；不接收外部数据库路径，不会打开生产数据。
    with TemporaryDirectory(prefix="sqlite-pagination-") as folder:
        path = Path(folder) / "records.sqlite3"
        conn = connect(str(path))
        try:
            seed(conn)
            yield conn, path
        finally:
            conn.close()

静态数据的三页分别是 106/105、104/103、102/101。对应的 has_more 为 true、true、false。最后一页刚好有两条，也不能只根据“返回数等于页大小”判断还有下一页；多取一条解决的正是这个歧义。

• 空列表：items 为空，has_more=false，next_cursor=None。
• 最后一页：可能不足一页，也可能刚好满页；没有探测行就结束。
• 到达末尾后：界面隐藏或禁用“加载更多”。不要把 next_cursor=None 又当成首次请求，否则会重新拉取首屏。
• 切换 feed、筛选条件或排序方式：清空旧列表和旧游标，从首屏开始。
• 请求重试或重复点击：界面按请求状态防重入，必要时按 id 合并结果；去重不能补回已经漏掉的记录。

has_more 只描述这一次 SELECT 看到的状态。它为 true 后，其他请求仍可能删除剩余记录，所以下一页可以为空；它为 false 后，也可能新插入记录。要看新内容，可以刷新首屏或设计明确的增量刷新入口。

仍从 106、105 这一页开始，把边界故意改错，观察失败特征：

-- 错误 A：包含边界行，105 会再次出现。
WHERE feed = ? AND (published, id) <= (?, ?)

-- 错误 B：丢掉破平局的 id，同秒的 104、103 都被排除。
WHERE feed = ? AND published < ?

在同一排序和页大小下，A 得到 105、104；B 得到 102、101。测试文件专门断言这两个错误结果，作为负对照。它们的测试通过，表示我们成功复现了已知错误，不表示这两种 SQL 可以用于正确分页。

删除也要分情况。首屏后删掉已读 106，OFFSET 第二页会变成 103、102，从而漏掉仍在库中的 104；复合游标仍返回 104、103。删掉游标所指的 105 后也一样，因为后续查询已经拿到了两个边界值，不必再查回 105。若删的是未读 104，下一页只会返回 103、102：分页方法不能恢复已经删除的数据。

复合边界能避开上面的位置漂移，但仍按每次查询时的记录值工作。首屏返回后，把未读的 104 从 10:01 改到 10:04，它就移到了边界之前；后续只能读到 103、102、101。反过来，把已经读过的 106 改到 09:58，它会落到边界之后，后续又出现一次。

即使旧记录的排序字段不变，补录到游标之后的新记录仍可能进入后续页。本例在首屏后插入发布时间为 10:00:30 的 108，后续结果是 104、103、108、102、101。这与“固定为首屏时刻的数据集合”不同。

SQLite 的隔离保证与事务边界有关。WAL 模式下，同一个显式读事务在首次读取后可以持续看到该快照；另一个连接之后提交的修改不会立刻进入这个读事务。结束事务再读取，视图才会前进。普通的独立分页请求不会因为用了游标就共享这个快照。SQLite：Isolation In SQLite

本例连接使用 autocommit=True，常规 list_page() 的结果通过 fetchall() 取完，不跨请求保留读事务。测试还用两个连接和 WAL 模式单独验证了显式 BEGIN、首次读取、另一连接写入、COMMIT 前后的差异。它用于说明边界，不建议让用户在网页上停留几分钟时一直占着数据库事务。

如果产品要求导出集合固定，可考虑在受控时间内完成一致性读取，或者先固定有序 ID 集合；如果连每条记录的内容也必须固定，还需要保留相应的数据版本或物化结果。高水位条件只能限制某些新记录进入，不能自动解决排序字段修改、删除或内容变更。先写清一致性需求，再决定存储与导出方案。

将下面内容保存为 demo.py，与前两个文件放在同一目录，然后运行 python -B demo.py。每个场景使用新建的合成数据库；打印结果方便你直接看到变化。它没有 Web 服务，也没有后台进程。

"""运行 python demo.py。输出固定场景结果，不做性能基准。"""
import json
import platform
import sqlite3
from fixtures import database, stamp
from pagination import Cursor, list_page


def ids(page):
    return [row.id for row in page.items]


def offset(conn):
    return [row[0] for row in conn.execute(
        "SELECT id FROM records WHERE feed = ? "
        "ORDER BY published DESC, id DESC LIMIT ? OFFSET ?",
        ("main", 2, 2),
    ).fetchall()]


def collect(conn, cursor):
    out = []
    while True:
        result = list_page(conn, "main", cursor)
        out.extend(ids(result))
        if not result.has_more:
            return out
        cursor = result.next_cursor


def main():
    results = {"runtime": {"python": platform.python_version(),
                           "sqlite": sqlite3.sqlite_version}}
    with database() as (conn, _):
        pages, cursor = [], None
        while True:
            page = list_page(conn, "main", cursor)
            pages.append({"ids": ids(page), "has_more": page.has_more})
            if not page.has_more:
                break
            cursor = page.next_cursor
        results["static_pages"] = pages
        first = list_page(conn, "main")
        boundary = first.next_cursor
        results["first_cursor"] = [boundary.published, boundary.id]
        results["wrong_inclusive"] = [r[0] for r in conn.execute(
            "SELECT id FROM records WHERE feed = ? AND (published, id) <= (?, ?) "
            "ORDER BY published DESC, id DESC LIMIT ?",
            ("main", boundary.published, boundary.id, 2),
        )]
        results["wrong_timestamp_only"] = [r[0] for r in conn.execute(
            "SELECT id FROM records WHERE feed = ? AND published < ? "
            "ORDER BY published DESC, id DESC LIMIT ?",
            ("main", boundary.published, 2),
        )]
        results["after_end"] = ids(list_page(conn, "main", Cursor("main", stamp(9, 59), 101)))
        results["empty_feed"] = ids(list_page(conn, "empty"))

    with database() as (conn, _):
        first = list_page(conn, "main")
        conn.execute("INSERT INTO records VALUES (?, ?, ?, ?)",
                     (107, "main", stamp(10, 3), "新顶部记录"))
        results["top_insert"] = {"offset": offset(conn),
                                 "keyset": ids(list_page(conn, "main", first.next_cursor))}

    with database() as (conn, _):
        first = list_page(conn, "main")
        conn.execute("DELETE FROM records WHERE id = ?", (106,))
        results["delete_seen"] = {"offset": offset(conn),
                                  "keyset": ids(list_page(conn, "main", first.next_cursor))}

    with database() as (conn, _):
        first = list_page(conn, "main")
        conn.execute("DELETE FROM records WHERE id = ?", (105,))
        results["delete_boundary"] = ids(list_page(conn, "main", first.next_cursor))

    with database() as (conn, _):
        first = list_page(conn, "main")
        conn.execute("UPDATE records SET published = ? WHERE id = ?", (stamp(10, 4), 104))
        results["unseen_moves_ahead"] = collect(conn, first.next_cursor)

    with database() as (conn, _):
        first = list_page(conn, "main")
        conn.execute("UPDATE records SET published = ? WHERE id = ?", (stamp(9, 58), 106))
        results["seen_moves_behind"] = collect(conn, first.next_cursor)

    with database() as (conn, _):
        first = list_page(conn, "main")
        conn.execute("INSERT INTO records VALUES (?, ?, ?, ?)",
                     (108, "main", stamp(10, 0, 30), "补录到游标之后"))
        results["backfill_appears"] = collect(conn, first.next_cursor)
    print(json.dumps(results, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

运行结果中的 Python 与 SQLite 版本取自当前解释器。下面列的是固定数据应出现的业务结果，版本号可不同：

除“翻三页”本身外，表格每行都从全新的 fixture 开始。后续结果表示从首屏保存的游标继续取到末尾，不包含首屏的 106、105。

将下面文件保存为 test_pagination.py，运行 python -B -m unittest -v test_pagination.py。除上面的手工场景外，还覆盖页大小为 1、恰好满页、部分末页、过滤范围不串列表、异常游标与页大小、空或非整数发布时间，以及独立连接和显式读事务的差异。

import sqlite3
import unittest
from fixtures import database, stamp
from pagination import Cursor, connect, list_page
from demo import collect, ids, offset


class PaginationTests(unittest.TestCase):
    def setUp(self):
        self.fixture = database()
        self.conn, self.path = self.fixture.__enter__()

    def tearDown(self):
        self.fixture.__exit__(None, None, None)

    def first(self):
        return list_page(self.conn, "main")

    def insert(self, record_id, timestamp):
        self.conn.execute("INSERT INTO records VALUES (?, ?, ?, ?)",
                          (record_id, "main", timestamp, "合成记录"))

    def test_static_all_pages_and_tie_crossing(self):
        first = self.first()
        second = list_page(self.conn, "main", first.next_cursor)
        third = list_page(self.conn, "main", second.next_cursor)
        self.assertEqual([ids(p) for p in (first, second, third)],
                         [[106, 105], [104, 103], [102, 101]])
        self.assertEqual([p.has_more for p in (first, second, third)],
                         [True, True, False])
        self.assertIsNone(third.next_cursor)
        self.assertEqual(first.next_cursor, Cursor("main", stamp(10, 1), 105))

    def test_extra_probe_row_is_not_the_cursor(self):
        # 若误用探测行 104 作游标，下一页会从 103 开始并漏掉 104。
        self.assertEqual(self.first().next_cursor.id, 105)

    def test_empty_feed(self):
        page = list_page(self.conn, "empty")
        self.assertEqual(ids(page), [])
        self.assertFalse(page.has_more)
        self.assertIsNone(page.next_cursor)

    def test_empty_after_last_record(self):
        page = list_page(self.conn, "main", Cursor("main", stamp(9, 59), 101))
        self.assertEqual(ids(page), [])
        self.assertFalse(page.has_more)
        self.assertIsNone(page.next_cursor)

    def test_exact_full_final_page(self):
        page = list_page(self.conn, "main", limit=6)
        self.assertEqual(ids(page), [106, 105, 104, 103, 102, 101])
        self.assertFalse(page.has_more)
        self.assertIsNone(page.next_cursor)

    def test_partial_final_page(self):
        first = list_page(self.conn, "main", limit=4)
        last = list_page(self.conn, "main", first.next_cursor, limit=4)
        self.assertEqual(ids(last), [102, 101])
        self.assertFalse(last.has_more)

    def test_page_size_one(self):
        cursor, found = None, []
        while True:
            page = list_page(self.conn, "main", cursor, limit=1)
            found.extend(ids(page))
            if not page.has_more:
                break
            cursor = page.next_cursor
        self.assertEqual(found, [106, 105, 104, 103, 102, 101])

    def test_top_insert_offset_repeats_keyset_continues(self):
        first = self.first()
        self.insert(107, stamp(10, 3))
        self.assertEqual(offset(self.conn), [105, 104])
        self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [104, 103])
        self.assertEqual(ids(self.first()), [107, 106])

    def test_delete_seen_offset_skips_keyset_continues(self):
        first = self.first()
        self.conn.execute("DELETE FROM records WHERE id = ?", (106,))
        self.assertEqual(offset(self.conn), [103, 102])
        self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [104, 103])

    def test_delete_boundary_cursor_needs_no_lookup(self):
        first = self.first()
        self.conn.execute("DELETE FROM records WHERE id = ?", (105,))
        self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [104, 103])

    def test_delete_unseen_is_not_recovered(self):
        first = self.first()
        self.conn.execute("DELETE FROM records WHERE id = ?", (104,))
        self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [103, 102])

    def test_inclusive_negative_control_repeats_boundary(self):
        c = self.first().next_cursor
        bad = [r[0] for r in self.conn.execute(
            "SELECT id FROM records WHERE feed = ? AND (published, id) <= (?, ?) "
            "ORDER BY published DESC, id DESC LIMIT ?", ("main", c.published, c.id, 2)
        ).fetchall()]
        self.assertEqual(bad, [105, 104])
        self.assertIn(105, bad)

    def test_timestamp_only_negative_control_misses_ties(self):
        c = self.first().next_cursor
        bad = [r[0] for r in self.conn.execute(
            "SELECT id FROM records WHERE feed = ? AND published < ? "
            "ORDER BY published DESC, id DESC LIMIT ?", ("main", c.published, 2)
        ).fetchall()]
        self.assertEqual(bad, [102, 101])
        self.assertNotIn(104, bad)
        self.assertNotIn(103, bad)

    def test_unseen_sort_key_moves_ahead_and_is_missed(self):
        first = self.first()
        self.conn.execute("UPDATE records SET published = ? WHERE id = ?", (stamp(10, 4), 104))
        self.assertEqual(collect(self.conn, first.next_cursor), [103, 102, 101])
        self.assertEqual(ids(self.first()), [104, 106])

    def test_seen_sort_key_moves_behind_and_reappears(self):
        first = self.first()
        self.conn.execute("UPDATE records SET published = ? WHERE id = ?", (stamp(9, 58), 106))
        rest = collect(self.conn, first.next_cursor)
        self.assertEqual(rest, [104, 103, 102, 101, 106])
        self.assertEqual((ids(first) + rest).count(106), 2)

    def test_backfilled_insert_appears_in_later_pages(self):
        first = self.first()
        self.insert(108, stamp(10, 0, 30))
        self.assertEqual(collect(self.conn, first.next_cursor), [104, 103, 108, 102, 101])

    def test_feed_is_filtered_and_bound(self):
        self.assertEqual(ids(list_page(self.conn, "other")), [999])
        self.assertEqual(ids(list_page(self.conn, "main' OR 1=1 --")), [])
        self.assertEqual(ids(self.first()), [106, 105])

    def test_cursor_scope_cannot_change(self):
        with self.assertRaises(ValueError):
            list_page(self.conn, "other", self.first().next_cursor)

    def test_bad_limits_and_cursor_values_rejected(self):
        for value in (0, -1, 101, True, 2.5, "2", None):
            with self.subTest(limit=value), self.assertRaises(ValueError):
                list_page(self.conn, "main", limit=value)
        for cursor in ("garbage", Cursor("main", None, 105),
                       Cursor("main", -1, 105), Cursor("main", 2**63, 105),
                       Cursor("main", True, 105), Cursor("main", stamp(10, 1), 0),
                       Cursor("main", stamp(10, 1), "105")):
            with self.subTest(cursor=cursor), self.assertRaises(ValueError):
                list_page(self.conn, "main", cursor)

    def test_null_or_non_integer_published_rejected(self):
        for value in (None, "not-a-date", 1.5, -1):
            with self.subTest(value=value), self.assertRaises(sqlite3.IntegrityError):
                self.insert(110, value)

    def test_independent_connections_have_no_cross_request_snapshot(self):
        first = self.first()
        self.assertFalse(self.conn.in_transaction)
        writer = connect(str(self.path))
        try:
            writer.execute("UPDATE records SET published = ? WHERE id = ?",
                           (stamp(10, 4), 104))
            self.assertFalse(writer.in_transaction)
            self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [103, 102])
        finally:
            writer.close()

    def test_wal_explicit_read_transaction_keeps_its_snapshot(self):
        # 只为说明隔离边界，真实网页不应为人类的翻页等待一直持有事务。
        mode = self.conn.execute("PRAGMA journal_mode=WAL").fetchone()[0]
        self.assertEqual(mode.lower(), "wal")
        writer = connect(str(self.path))
        try:
            self.conn.execute("BEGIN")
            first = self.first()  # 首次读取建立快照。
            writer.execute("UPDATE records SET published = ? WHERE id = ?",
                           (stamp(10, 4), 104))
            self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [104, 103])
            self.conn.execute("COMMIT")
            self.assertEqual(ids(list_page(self.conn, "main", first.next_cursor)), [103, 102])
        finally:
            if self.conn.in_transaction:
                self.conn.execute("ROLLBACK")
            writer.close()


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

测试全部使用真实 sqlite3 查询，没有伪造数据库返回值。WAL 测试的 journal_mode 必须真的返回 wal；如果运行环境不支持它，测试应明确失败，不能把该隔离场景算作通过。本次普通运行和本地示例包解压到另一临时目录后的运行都通过，未进行负载测试或性能基准。

按文件名逐个复制上面四份完整源码，放到同一个 example 目录即可运行。本文不依赖额外下载。运行命令如下：

# 在 example 目录运行；不需要安装第三方包。
python -B demo.py
python -B -m unittest -v test_pagination.py

前面的分页函数在 Python 内部传递 Cursor，整数没有经过 JavaScript。接成 JSON 接口时还多一道边界：SQLite 和本例允许 id 达到 9223372036854775807，而 JavaScript Number 的安全整数上限是 9007199254740991。超过安全范围不表示每个整数都会变，但相邻整数不再保证能区分。RFC 8259 第 6 节说明了 JSON 数字在不同实现间的精度限制。

例如，服务端原本发出数字 9007199254740993，客户端 JSON.parse 后再 JSON.stringify，回传的是 9007199254740992。SQL 仍然使用严格小于，输入也仍然是合法整数，却已经拿到了另一个边界。换成 9007199254740995，会舍入为 9007199254740996。这两个方向分别能造成漏行和重行，不能仅用“能解析、没报错”验收。

本补充采用一个明确的传输约定：id 与 published 一律用十进制字符串，v 固定为数字 1，feed 继续绑定服务端已经允许访问的列表。published 的常用秒数远小于安全上限，但原函数允许完整 64 位范围；两列统一编码，避免同一个协议有两套精度规则。示例对象如下：

{"v":1,"feed":"main","published":"100","id":"9007199254740993"}

保存下面的 cursor_wire.py，与原来的四个 Python 文件放在一起。它只负责已解码对象与内部 Cursor 的互换，不改 SQL、数据库表或原来的分页函数。允许的数字字符串只有 ASCII 数字，无正负号、空白、小数、指数或多余前导零；在转 int 前先检查长度，再检查原来的整数范围。

"""Internal Cursor <-> JSON object. Decimal strings preserve 64-bit integers.
This is a value codec, not authentication, a signed token, or a raw-JSON parser.
"""
import re
from pagination import Cursor

MAX_I64 = 2**63 - 1
FIELDS = {"v", "feed", "published", "id"}


def _feed(value):
    if type(value) is not str or not 1 <= len(value) <= 128:
        raise ValueError("invalid feed")


def _native_integer(value, minimum):
    if type(value) is not int or not minimum <= value <= MAX_I64:
        raise ValueError("invalid integer")


def _decimal(value, minimum):
    if (type(value) is not str or not 1 <= len(value) <= 19
            or re.fullmatch(r"0|[1-9][0-9]*", value) is None):
        raise ValueError("expected canonical ASCII decimal string")
    number = int(value)
    _native_integer(number, minimum)
    return number


def cursor_to_wire(cursor):
    if type(cursor) is not Cursor:
        raise ValueError("expected internal Cursor")
    _feed(cursor.feed)
    _native_integer(cursor.published, 0)
    _native_integer(cursor.id, 1)
    return {"v": 1, "feed": cursor.feed,
            "published": str(cursor.published), "id": str(cursor.id)}


def cursor_from_wire(value, expected_feed):
    # expected_feed must come from the server's authorized request context.
    _feed(expected_feed)
    if type(value) is not dict or set(value) != FIELDS:
        raise ValueError("unexpected cursor fields")
    if type(value["v"]) is not int or value["v"] != 1:
        raise ValueError("unsupported cursor version")
    _feed(value["feed"])
    if value["feed"] != expected_feed:
        raise ValueError("cursor belongs to another feed")
    return Cursor(expected_feed, _decimal(value["published"], 0),
                  _decimal(value["id"], 1))

客户端保存并原样回传这两个字符串，不要再用 Number、parseInt 或一元加号转回数字。服务端验证成功后才还原为 Python int，交给原来的参数绑定查询。字符串只是传输格式，不应按字符串的字典顺序重新排列记录。

这里检查 feed 是为了防止游标串列表，不能授予访问权限。expected_feed 必须来自服务端已鉴权的请求上下文，不能直接照抄客户端游标。这个对象没有签名、加密、过期校验或防重放能力，也没有实现 URL token。

保存下面的 wire_demo.py。它会新建临时合成数据库，替换其中的演示数据，再调用本机已有的 Node 执行 JSON.parse / JSON.stringify。不会连接线上数据库、启动 Web 服务或调用 Codex。需要原文所述 Python 3.12+，以及能从命令行运行的 Node；本次实测 Node v24.19.0，没有据此声明其他版本的完整兼容性。

每轮放四条同 published 的记录，先取两条，把第二条作为游标。数字分支故意采用不安全的 JSON 数字协议，字符串分支使用上面的编解码函数。两分支分别与没有经过客户端的精确边界比较。

"""Owned temporary SQLite only. Requires Python 3.12+ and an existing Node.
Runs real Node JSON.parse/stringify, not a browser or a web API.
"""
import json
import platform
import sqlite3
import subprocess
from fixtures import database
from pagination import Cursor, list_page
from cursor_wire import cursor_from_wire, cursor_to_wire

JS_ROUND_TRIP = """
const fs = require('node:fs');
const value = JSON.parse(fs.readFileSync(0, 'utf8'));
process.stdout.write(JSON.stringify(value));
"""


def node_round_trip(value):
    result = subprocess.run(
        ["node", "-e", JS_ROUND_TRIP], input=json.dumps(value),
        text=True, capture_output=True, check=True, timeout=5,
    )
    return json.loads(result.stdout)


def ids(page):
    return [row.id for row in page.items]


def run_case(boundary):
    with database() as (conn, _):
        conn.execute("DELETE FROM records")  # Newly created synthetic fixture.
        conn.executemany("INSERT INTO records VALUES (?, ?, ?, ?)",
                         [(n, "main", 100, "synthetic")
                          for n in (boundary + 1, boundary, boundary - 1, boundary - 2)])
        first = list_page(conn, "main", limit=2)
        original = first.next_cursor
        exact = ids(list_page(conn, "main", original))
        unsafe = node_round_trip({"feed": original.feed,
                                  "published": original.published, "id": original.id})
        rounded = Cursor(unsafe["feed"], unsafe["published"], unsafe["id"])
        damaged = ids(list_page(conn, "main", rounded))
        safe = cursor_from_wire(node_round_trip(cursor_to_wire(original)), "main")
        preserved = ids(list_page(conn, "main", safe))
        if exact != [boundary - 1, boundary - 2] or preserved != exact:
            raise AssertionError("safe boundary changed")
        if safe != original or damaged == exact:
            raise AssertionError("negative control did not expose precision loss")
        return {"boundary": str(boundary), "numeric_round_trip": str(rounded.id),
                "expected_next": list(map(str, exact)),
                "numeric_next": list(map(str, damaged)),
                "string_next": list(map(str, preserved))}


def main():
    result = {"runtime": {"python": platform.python_version(),
                          "sqlite": sqlite3.sqlite_version,
                          "node": subprocess.run(["node", "--version"], text=True,
                                                 capture_output=True, check=True,
                                                 timeout=5).stdout.strip()},
              "round_down": run_case(2**53 + 1),
              "round_up": run_case(2**53 + 3)}
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

python -B wire_demo.py

两个字符串分支都返回表中“应返回的下一页”。这里没有并发写入，也没有改排序键，问题仅来自游标经过数字解码。输出中的 id 也刻意用字符串，避免把证据再交给另一个数字解析器时重新丢失精度。

更大的值还可能被舍入到允许范围以外，从而在服务端报错，并不一定表现为静默漏读。不要把上述两个固定结果推广成“所有大整数都按同一个方向变化”。

保存 test_cursor_wire.py。新增的 10 项 unittest 方法覆盖两个实际分页反例、五组大整数的真实 Node 字符串往返、最小/最大边界、错误字段集合、版本与 feed、不合法数字字符串和 Python 内部值。一个方法中的多组输入不是额外的测试方法。最大 published 只测试约定的整数范围，不把它解释为可用的真实日期。

import json
import unittest
from pagination import Cursor
from cursor_wire import MAX_I64, cursor_from_wire, cursor_to_wire
from wire_demo import node_round_trip, run_case


class WireTests(unittest.TestCase):
    def valid(self):
        return {"v": 1, "feed": "main", "published": "100", "id": "9007199254740993"}

    def test_minimum_and_maximum_round_trip(self):
        for published, record_id in ((0, 1), (MAX_I64, MAX_I64)):
            cursor = Cursor("main", published, record_id)
            wire = cursor_to_wire(cursor)
            self.assertIs(type(wire["id"]), str)
            self.assertIs(type(wire["published"]), str)
            self.assertEqual(cursor_from_wire(json.loads(json.dumps(wire)), "main"), cursor)

    def test_real_node_string_round_trip_including_i64_max(self):
        for value in (2**53 - 1, 2**53, 2**53 + 1, 2**53 + 3, MAX_I64):
            cursor = Cursor("main", value, value)
            self.assertEqual(cursor_from_wire(node_round_trip(cursor_to_wire(cursor)), "main"), cursor)

    def test_round_down_misses_still_present_row(self):
        result = run_case(2**53 + 1)
        self.assertEqual(result["numeric_round_trip"], "9007199254740992")
        self.assertEqual(result["numeric_next"], ["9007199254740991"])
        self.assertEqual(result["string_next"], ["9007199254740992", "9007199254740991"])

    def test_round_up_repeats_boundary(self):
        result = run_case(2**53 + 3)
        self.assertEqual(result["numeric_round_trip"], "9007199254740996")
        self.assertEqual(result["numeric_next"], ["9007199254740995", "9007199254740994"])
        self.assertEqual(result["string_next"], ["9007199254740994", "9007199254740993"])

    def test_noncanonical_decimal_rejected(self):
        for field in ("id", "published"):
            for value in ("", "01", "+1", "-1", " 1", "1 ", "1\n", "1.0", "1e3", "１", "١", "9" * 20):
                wire = self.valid(); wire[field] = value
                with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                    cursor_from_wire(wire, "main")

    def test_numeric_bool_null_and_container_rejected(self):
        for field in ("id", "published"):
            for value in (1, 1.0, True, None, [], {}):
                wire = self.valid(); wire[field] = value
                with self.subTest(field=field, value=value), self.assertRaises(ValueError):
                    cursor_from_wire(wire, "main")

    def test_out_of_range_rejected(self):
        for field, value in (("id", "0"), ("id", str(MAX_I64 + 1)), ("published", str(MAX_I64 + 1))):
            wire = self.valid(); wire[field] = value
            with self.subTest(field=field), self.assertRaises(ValueError):
                cursor_from_wire(wire, "main")

    def test_exact_field_set_and_object_required(self):
        values = [None, [], "cursor", {**self.valid(), "extra": 1}]
        for field in self.valid():
            wire = self.valid(); del wire[field]; values.append(wire)
        for wire in values:
            with self.subTest(value=wire), self.assertRaises(ValueError):
                cursor_from_wire(wire, "main")

    def test_version_and_feed_context_rejected(self):
        for value in (True, 1.0, "1", 2, None):
            wire = self.valid(); wire["v"] = value
            with self.subTest(version=value), self.assertRaises(ValueError):
                cursor_from_wire(wire, "main")
        for value in (None, "", "x" * 129, "other"):
            wire = self.valid(); wire["feed"] = value
            with self.subTest(feed=value), self.assertRaises(ValueError):
                cursor_from_wire(wire, "main")
        with self.assertRaises(ValueError):
            cursor_from_wire(self.valid(), None)

    def test_invalid_internal_cursor_not_silently_coerced(self):
        for value in (None, self.valid(), Cursor("", 100, 1), Cursor("main", True, 1),
                      Cursor("main", 100, True), Cursor("main", -1, 1),
                      Cursor("main", 100, 0), Cursor("main", 100, MAX_I64 + 1)):
            with self.subTest(value=value), self.assertRaises(ValueError):
                cursor_to_wire(value)


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

# 同一目录中保留原文四个文件，再加入本补充三个文件。
python -B -m unittest -v test_pagination.py test_cursor_wire.py
python -B -O -m unittest -v test_pagination.py test_cursor_wire.py

2026-10-20 在 Python 3.12.14 / SQLite 3.53.1 / Node v24.19.0 上运行：原文 22 项和新增 10 项，共 32 项，在普通模式与 -O 模式分别通过。两次运行不相加成 64 项。原文 2026-10-13 的测试日期、环境记录和图解保持不变；本补充是新增的本地跨运行时实验。

编解码函数接收的是已经解析出的字典。原始请求体的字节上限、嵌套深度、重复 JSON 键处理仍属于接口解析层；不能把这里的四字段检查当成完整的请求防护。Python 默认 json.loads 会保留重复键的最后一个值，解析后这个函数无法知道之前出现过重复键。Python JSON：重复键行为。本文 demo 的 JSON 由本地程序自己生成，不代表验证了不可信 HTTP 请求。

另外，若响应 records 中的 id 仍然是 JSON 数字，客户端的记录去重、选择状态或缓存键仍可能合并不同的记录。游标字符串化只修复游标这一段；实际接口应同时审查记录标识的传输格式，并测试真实前端的往返路径。没有验证浏览器、Web 框架、生产接口、签名 token、鉴权或真实 Codex 执行，也没有改站点功能控件。

1. 确认业务顺序。选非空排序字段，使用唯一且稳定的第二键；明确 published 是否允许修改。
2. 让第一页与后续页保持相同过滤条件。服务端先鉴权，再决定允许查询的 feed；游标里的 feed 只能用于一致性检查，不能授权访问。
3. 若要生成客户端 token，设计版本、过滤上下文、排序方向、字段类型、长度限制和过期策略。需要防篡改时再加入签名或服务端状态；Base64 本身不保密，也不防篡改。
4. 前端按 has_more 收尾，切换筛选时清空旧游标。不要继续展示无法兑现的固定总页数或任意跳页入口。
5. 将本例的同秒、插入、删除和排序键更新场景迁入项目测试，再核对真实索引和查询计划。索引调整应走项目现有迁移流程。

交给 Codex 修改项目时，可以用下面这段任务说明。它是一个可复用的任务模板，本文没有声称在任何真实业务仓库中执行过它。

先阅读当前记录列表查询、接口返回值、前端加载更多逻辑和已有测试。
将该列表改为 published DESC, id DESC 的向后游标分页，保留现有权限与过滤条件。
所有动态值都绑定参数；多取一行检测 has_more，游标取最后一条实际返回记录。
不要连接或修改生产数据库，不要顺带重构无关功能。

先写失败用例，再修改：
1. 相同 published 的记录跨页，不能缺少或重复。
2. 首屏后顶部插入，后续页不能因 OFFSET 位置移动而重复。
3. 删除已读行或游标行后可继续；删除未读行不承诺恢复。
4. <= 和仅时间游标分别能复现边界重复与同秒遗漏。
5. 空页、整页末尾、异常游标、切换过滤范围均有明确结果。
6. 修改排序键能展示游标不是快照，不能写成跨请求一致性保证。

运行项目原有测试及新增测试，说明实际环境、通过范围、未验证项。
如需改接口契约或索引迁移，先列出影响，再按项目流程处理。

这次工作是否完成，应以旧接口行为被明确替换、客户端结束条件正确、负例可复现且回归测试通过来判断。把 SQL 里的 OFFSET 删除，只是其中一步。

• SQLite Row Values：滚动窗口查询，用于核对多列边界比较思路。
• SQLite Row Values：版本兼容，行值语法从 3.15.0 起提供。
• SQLite SELECT：ORDER BY，用于核对排序与相同键的顺序边界。
• SQLite：隔离行为，用于核对 WAL 下显式读事务和跨连接可见性。
• Python sqlite3.connect 与 参数绑定说明，用于核对连接设置与占位符。

2026-10-13 实测：Python 3.12.14 / SQLite 3.53.1，22 项 unittest 通过；本地示例包在独立临时目录解压后重跑通过。图解为本教程原创 SVG，已离线渲染检查，属于合成数据结构图。完整浏览器页面与窄屏展示尚未实测；这里只完成了 HTML、嵌入源码、图像尺寸和素材一致性检查。没有生产数据实验、性能收益承诺、真实产品截图或外部模型运行记录。