Flask如何实现百万级数据的分页查询_Python基于游标分页与SQLAlchemy优化-Python教程-PHP中文网

offset/limit在百万数据下变慢是因为数据库需扫描并丢弃前n行，如查第1000页（每页20条）要扫描2万行；游标分页通过唯一有序字段（如id）条件过滤，复杂度稳定o(log n)，但需索引支持且不支持跳页。

flask如何实现百万级数据的分页查询_python基于游标分页与sqlalchemy优化

为什么 offset/limit 在百万数据下会变慢

因为 OFFSET 要让数据库先跳过前 N 行再取结果，MySQL 或 PostgreSQL 都得扫描并丢弃这些行。100 万条数据查第 1000 页（每页 20 条），就得扫描 2 万行——不是只查 20 行。越往后翻页，越慢，还容易锁表。

游标分页绕开 OFFSET，靠上一页最后一条记录的唯一、有序字段（比如 id 或 created_at）做条件过滤，每次只查“比它大/小的下一批”，复杂度稳定在 O(log n)。

必须有索引支撑：用于游标的字段（如 id）要建索引，复合排序字段也要联合索引（例如 (status, id)）
不能跳页：用户只能“下一页”或“上一页”，不支持直接跳转到第 87 页
数据实时性要求高时需注意：新插入记录可能插在中间，导致漏数或重复（后文会提应对）

SQLAlchemy 中实现基于 id 的游标分页

假设模型是 User，主键 id 自增且不删除，这是最简场景：

def get_users_after(cursor_id: int, limit: int = 20):
    return db.session.query(User).filter(User.id > cursor_id).order_by(User.id).limit(limit).all()

返回结果后，取最后一项的 id 作为下一页的 cursor_id。注意三点：

必须用 >（升序）或 （降序），不能用 <code>>=，否则可能重复
order_by() 字段必须和游标字段一致，且方向固定；若按 created_at 排序，需确保该字段唯一或加 id 作二级排序
首次请求没有 cursor_id，传 0 或 None 并特殊处理（如用 filter(User.id > 0)）

处理非自增/非唯一字段的游标（如 created_at）

时间戳常有重复，直接用 created_at > last_time 会漏掉同秒插入的多条记录。稳妥做法是组合主键：

python-parallelization

将顺序 Python 代码转换为并行/并发实现，用于在需要并行化 Python 代码、提升性能时使用。

下载

# 假设 last_item = { 'created_at': '2024-05-01 10:00:00', 'id': 12345 }
query = db.session.query(User).filter(
    (User.created_at > last_item['created_at']) | 
    ((User.created_at == last_item['created_at']) & (User.id > last_item['id']))
).order_by(User.created_at, User.id).limit(20)

这个写法保证严格单调，但要注意：

WHERE 条件中用了 OR，数据库可能无法高效走索引——必须建联合索引 (created_at, id)
如果业务允许，更推荐把 created_at 存为带微秒的 DateTime 或改用 UUIDv7 类时间有序 ID，减少冲突
前端传游标时，应传 base64 编码后的复合值（如 base64.b64encode(b'2024-05-01 10:00:00|12345')），避免 URL 中特殊字符或类型丢失

Flask 路由与响应结构设计

不要在 URL 里暴露原始字段值，用语义化参数名，并校验游标有效性：

@app.route('/api/users')
def list_users():
    cursor = request.args.get('cursor')
    limit = min(100, int(request.args.get('limit', 20)))  # 防爆破
    if not cursor:
        items = get_users_after(0, limit)
        next_cursor = items[-1].id if items else None
    else:
        try:
            cursor_id = int(cursor)
            items = get_users_after(cursor_id, limit)
            next_cursor = items[-1].id if items else None
        except (ValueError, IndexError):
            return {'error': 'invalid cursor'}, 400
    return {
        'data': [u.to_dict() for u in items],
        'next': next_cursor,
        'has_more': len(items) == limit
    }

关键点：