FastAPI+Redis分页场景：实时数据更新后的API内容同步方案咨询

针对你的场景（Redis列表实时增删，FastAPI分页接口需返回最新数据），可以从以下几个方面调整API：

1. 每次请求直接基于Redis最新数据计算分页 #

放弃任何本地缓存，接口处理请求时直接从Redis获取当前列表的最新状态，再进行分页计算：

• 先通过LLEN命令获取列表当前的总长度，以此计算分页的起始/结束索引（Redis列表是0-based索引）；
• 再用LRANGE命令拉取对应索引范围的最新数据。

示例代码：

from fastapi import FastAPI, Query
import redis

app = FastAPI()
# 初始化Redis集群连接（根据实际集群配置调整）
redis_cluster = redis.RedisCluster(host="your-cluster-host", port=your-port, decode_responses=True)

@app.get("/cached-data/{key}")
def get_paginated_data(
    key: str,
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1)
):
    # 获取当前列表总长度（最新值）
    total_count = redis_cluster.llen(key)
    if total_count == 0:
        return {"data": [], "total": 0, "page": page, "size": size}
    
    # 计算分页索引
    start_idx = (page - 1) * size
    end_idx = start_idx + size - 1
    # 处理超出列表长度的情况
    if end_idx >= total_count:
        end_idx = total_count - 1
    # 拉取最新分页数据
    paginated_data = redis_cluster.lrange(key, start_idx, end_idx)
    
    return {
        "data": paginated_data,
        "total": total_count,
        "page": page,
        "size": size
    }

2. 禁用接口层的所有缓存策略 #

检查FastAPI接口是否使用了@lru_cache这类本地缓存装饰器，或者反向代理（如Nginx）是否对该接口开启了缓存。这类缓存会留存旧的分页结果，必须禁用：

• 移除接口上的@lru_cache等缓存装饰器；
• 配置反向代理时，对该接口设置no-cache响应头，避免客户端或代理缓存结果。

示例：给接口添加禁用缓存的响应头

from fastapi import Response

@app.get("/cached-data/{key}")
def get_paginated_data(key: str, page: int = Query(1, ge=1), size: int = Query(10, ge=1), response: Response = None):
    # 设置响应头禁用缓存
    response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
    response.headers["Pragma"] = "no-cache"
    response.headers["Expires"] = "0"
    
    # 后续分页逻辑同前
    total_count = redis_cluster.llen(key)
    # ...

3. 用Redis原子操作避免分页计算时数据变更 #

如果在LLEN获取长度和LRANGE拉取数据的间隙，列表被其他服务修改，会导致分页结果不准确。可以用Redis Lua脚本将这两个操作封装成原子操作，避免中间数据变化：

示例Lua脚本及调用：

# 定义Lua脚本，原子执行长度查询和分页数据拉取
pagination_script = """
local target_key = ARGV[1]
local page_num = tonumber(ARGV[2])
local page_size = tonumber(ARGV[3])

local total = redis.call('LLEN', target_key)
local start = (page_num - 1) * page_size
local end_idx = start + page_size - 1

if end_idx >= total then
    end_idx = total - 1
end

if start > end_idx then
    return {total, {}}
end

local data = redis.call('LRANGE', target_key, start, end_idx)
return {total, data}
"""

@app.get("/cached-data/{key}")
def get_paginated_data(
    key: str,
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1)
):
    # 执行Lua脚本
    result = redis_cluster.eval(pagination_script, 0, key, page, size)
    total_count = result[0]
    paginated_data = result[1]
    
    return {
        "data": paginated_data,
        "total": total_count,
        "page": page,
        "size": size
    }

4. 可选：添加版本标识辅助客户端验证 #

可以给每个Redis列表维护一个版本号，每次修改数据时递增版本号，接口返回时带上该版本号，客户端可通过版本号判断是否拿到了最新数据：

• 修改数据的服务中，每次操作列表后调用HINCRBY递增对应key的版本；
• 分页接口中获取该版本号并返回。

示例：

# 更新数据的服务逻辑（示例）
def update_redis_list(key: str, new_value: str):
    # 添加新值
    redis_cluster.rpush(key, new_value)
    # 递增版本号
    redis_cluster.hincrby("data_versions", key, 1)

# 分页接口返回版本号
@app.get("/cached-data/{key}")
def get_paginated_data(key: str, page: int = Query(1, ge=1), size: int = Query(10, ge=1)):
    # ... 分页逻辑 ...
    # 获取当前版本号
    current_version = redis_cluster.hget("data_versions", key)
    current_version = int(current_version) if current_version else 0
    
    return {
        "data": paginated_data,
        "total": total_count,
        "page": page,
        "size": size,
        "version": current_version
    }

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