如何通过分页从Firebase Realtime DB获取子节点键

我需要从Firebase Realtime DB中获取子节点键，数据结构示例如下：

persons
    p1
        first_name: John
        surname: Blue
    p2
        first_name: Johnny
        surname: Red
    p3
        first_name: Mike
        surname: Orange

目标是得到键列表 [p1, p2, p3]，但实际节点数量超过10K（比如[p1, p2, ..., p11000]）。

我原本用以下代码实现：

def get_keys_only(db_path):
    ref = db.reference(db_path) 

    # first arg must be False for shallow queries
    # ref.get(includeEtag, shallowQuery)
    valueAtRef_as_dict = ref.get(False, True)

    # The returned data can be None if the collection doesn't exist in DB
    if valueAtRef_as_dict is not None:
        # Keys returned in a list
        return_list = [*valueAtRef_as_dict]
    else:
        return_list = []

    return return_list

但运行时抛出firebase_admin.exceptions.UnavailableError，异常详情：

in get_keys_only valueAtRef_as_dict = ref.get(False, True) File "/usr/local
/lib/python3.8/site-packages/firebase_admin/db.py", line 221, in get return 
self._client.body('get', self._add_suffix(), params=params) 
File "/usr/local/lib/python3.8/site-packages/firebase_admin/_http_client.py", line 131, 
in body resp = self.request(method, url, **kwargs) 
File "/usr/local/lib/python3.8/site-packages/firebase_admin/db.py", 
line 931, in request raise _Client.handle_rtdb_error(error) 
firebase_admin.exceptions.UnavailableError: Request has timed out

测试发现小数据量时代码正常，推测需要分页实现，但不知道怎么操作。请问如何通过分页获取大量键？如果分页不可行，还有其他解决办法吗？

一、分页实现方案 #

Firebase Realtime DB支持通过orderByKey()结合startAt()、limitToFirst()进行分页查询，配合shallow模式只获取键。具体实现代码如下：

def get_keys_paginated(db_path, page_size=1000):
    ref = db.reference(db_path)
    all_keys = []
    last_key = None

    while True:
        query = ref.order_by_key()
        # 若存在上一页最后一个键，以此为起点查询下一页
        if last_key:
            # 多取1条数据，用来判断是否还有后续分页
            query = query.start_at(last_key).limit_to_first(page_size + 1)
        else:
            query = query.limit_to_first(page_size)
        
        # 启用shallow模式，仅返回键的字典结构
        result = query.get(False, True)
        if not result:
            break
        
        current_keys = list(result.keys())
        # 若返回数量等于page_size+1，说明还有下一页，去掉最后一条并更新last_key
        if len(current_keys) == page_size + 1:
            current_keys = current_keys[:-1]
            last_key = current_keys[-1]
        else:
            last_key = None
        
        all_keys.extend(current_keys)
        # 无后续分页则退出循环
        if not last_key:
            break
    
    return all_keys

说明 #

• page_size可根据网络情况调整（比如500或1000），避免单次请求数据量过大导致超时
• 多取1条数据的目的是判断是否还有后续分页，确保所有键都能被获取
• 始终保持shallow模式，仅传输键数据，减少网络负载

二、替代解决方案 #

如果分页仍遇到超时问题，可尝试以下方案：

1. 调整SDK超时时间 #

通过自定义HTTP客户端配置，延长请求超时阈值：

from firebase_admin import db
from firebase_admin._http_client import HttpClient

# 设置超时时间为30秒（可按需调整）
custom_http_client = HttpClient(timeout=30)
# 重新初始化DB客户端，使用自定义配置
db._initialize_app(你的DB_APP实例, {'databaseURL': '你的数据库URL'}, http_client=custom_http_client)

2. 预优化数据结构 #

在写入数据时同步维护一个专门存储键的节点，比如person_keys：

persons
    p1: {...}
    p2: {...}
person_keys
    p1: true
    p2: true

后续直接读取person_keys节点的键即可，shallow查询小节点不会触发超时。

3. 直接调用REST API并添加重试机制 #

使用requests库直接调用Firebase REST API，配合分页参数和重试逻辑：

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def get_keys_via_rest(db_url, path, page_size=1000):
    session = requests.Session()
    # 添加重试机制，针对超时等异常自动重试
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    
    all_keys = []
    last_key = None
    base_url = f"{db_url}/{path}.json?shallow=true&orderBy=%22%24key%22"

    while True:
        url = base_url
        if last_key:
            url += f"&startAt=%22{last_key}%22&limitToFirst={page_size + 1}"
        else:
            url += f"&limitToFirst={page_size}"
        
        response = session.get(url)
        response.raise_for_status()
        result = response.json()
        
        if not result:
            break
        
        current_keys = list(result.keys())
        if len(current_keys) == page_size + 1:
            current_keys = current_keys[:-1]
            last_key = current_keys[-1]
        else:
            last_key = None
        
        all_keys.extend(current_keys)
        if not last_key:
            break
    
    return all_keys

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