FastAPI + CosmosDB场景下基于Python SDK按指定页码获取分页数据

Cosmos DB 底层采用延续令牌（Continuation Token）实现分页，本身不支持原生的 OFFSET 分页语法，无法直接通过指定页码一步查询到对应数据，你贴的官方示例已经展示了延续令牌的核心用法，我们可以基于该能力做二次封装实现按页码查询。

方案1：小数据量场景直接遍历到目标页码 #

适合页码数值小、分页查询频率低的场景，实现最简单，不需要额外存储。

from fastapi import FastAPI
from azure.cosmos import CosmosClient

app = FastAPI()
# 初始化Cosmos客户端，参数替换为自身配置即可
cosmos_client = CosmosClient("<你的Cosmos DB服务端点>", "<你的访问密钥>")
db = cosmos_client.get_database_client("<数据库名称>")
container = db.get_container_client("<容器名称>")

def get_page_by_num(partition_key: str, query: str, page_size: int, target_page: int):
    # 构造查询迭代器，指定单页返回数据量
    query_iterable = container.query_items(
        query=query,
        partition_key=partition_key,
        max_item_count=page_size
    )
    pager = query_iterable.by_page()
    current_page = 1
    target_page_items = []
    # 逐页遍历直到到达目标页码
    while current_page <= target_page:
        try:
            current_page_items = list(pager.next())
            # 已经遍历完所有数据仍未到目标页码，返回空
            if not current_page_items:
                return []
            target_page_items = current_page_items
            current_page += 1
        except StopIteration:
            return []
    return target_page_items

# FastAPI接口调用示例
@app.get("/items")
def list_items(page: int = 1, page_size: int = 10, partition_key: str = "pk"):
    query = "SELECT * FROM c"
    items = get_page_by_num(partition_key, query, page_size, page)
    return {"items": items, "page": page, "page_size": page_size}

• 优点：实现逻辑简单无额外依赖，不需要额外存储资源
• 缺点：页码越大查询耗时越长，需要从第一页开始逐页遍历到目标页，不适合大页码、高频查询的场景

方案2：生产级方案：缓存页码与延续令牌映射 #

适合高频分页、大页码场景，核心是把每次查询得到的延续令牌和对应页码绑定缓存，后续查询同条件的指定页码时直接取缓存的延续令牌查询，不需要从头遍历。
以下是基于Redis做缓存的实现示例：

import redis
redis_client = redis.Redis(host="localhost", port=6379, db=0)

def get_page_by_num_with_cache(partition_key: str, query: str, page_size: int, target_page: int):
    # 生成唯一缓存键，需把所有影响查询结果的条件都拼接进去
    cache_key = f"cosmos_page:{hash(f'{partition_key}{query}{page_size}')}"
    # 先查缓存有没有目标页码前一页的延续令牌
    continuation_token = redis_client.get(f"{cache_key}:{target_page-1}")
    query_iterable = container.query_items(
        query=query,
        partition_key=partition_key,
        max_item_count=page_size
    )
    # 命中缓存直接用令牌取对应页
    if continuation_token:
        pager = query_iterable.by_page(continuation_token.decode())
        return list(pager.next())
    # 未命中缓存则逐页遍历，同时写入缓存
    pager = query_iterable.by_page()
    current_page = 1
    target_page_items = []
    while current_page <= target_page:
        try:
            current_page_items = list(pager.next())
            if not current_page_items:
                return []
            # 缓存当前页对应的下一页延续令牌，有效期1小时可自行调整
            if pager.continuation_token:
                redis_client.setex(f"{cache_key}:{current_page}", 3600, pager.continuation_token)
            target_page_items = current_page_items
            current_page += 1
        except StopIteration:
            return []
    return target_page_items

注意事项：

• 数据有新增/删除/更新时，需要清空对应查询条件的缓存，避免页码和数据对应错乱
• 缓存过期时间可根据业务数据的更新频率灵活调整

内容的提问来源于stack exchange，提问作者SLN