Python如何将Cloud Firestore返回的对象数组映射为对象列表

目标处理对象：training_dataset集合下所有文档内的ingredients数组字段

前置准备 #

• 安装Firebase官方Python SDK：执行命令 pip install firebase-admin
• 准备好Firebase项目的服务账号密钥文件，在Firebase控制台项目设置的服务账号板块可以生成

基础连接初始化 #

先完成Firestore客户端的初始化，后续所有操作都基于这个客户端实例：

import firebase_admin
from firebase_admin import credentials, firestore

# 替换为你本地存放服务账号密钥的实际路径
cred = credentials.Certificate("./firebase-service-key.json")
firebase_admin.initialize_app(cred)
db = firestore.client()

核心映射逻辑 #

Firestore拉取到的数组字段会自动转为Python原生list类型，不需要额外做格式解析，直接遍历处理即可。

小数据量全量映射 #

如果集合文档总数在1000条以内，可以直接全量拉取映射：

# 指向目标集合
training_docs = db.collection("training_dataset").stream()
result_map = {}

for doc in training_docs:
    doc_data = doc.to_dict()
    # 跳过没有ingredients字段的异常文档
    if "ingredients" not in doc_data:
        continue
    # 以文档ID为键，对应ingredients数组为值存储
    result_map[doc.id] = doc_data["ingredients"]

    # 如果需要对数组内元素做自定义映射，比如提取原料名、统一格式，可以直接遍历数组处理
    # 示例：数组元素为对象时，提取每个原料的name字段，过滤无效值
    # result_map[doc.id] = [
    #     item["name"].strip() 
    #     for item in doc_data["ingredients"] 
    #     if isinstance(item, dict) and "name" in item
    # ]

大数据量分页映射 #

如果集合文档量较大，建议分页拉取避免内存占用过高、请求超时：

PAGE_SIZE = 100  # 单页拉取数量，最大可设为1000
collection_ref = db.collection("training_dataset")
result_map = {}
last_doc = None

while True:
    # 构造分页查询
    if last_doc:
        page_query = collection_ref.start_after(last_doc).limit(PAGE_SIZE)
    else:
        page_query = collection_ref.limit(PAGE_SIZE)
    
    current_page = list(page_query.stream())
    # 拉取完所有文档时退出循环
    if not current_page:
        break

    # 处理当前页数据
    for doc in current_page:
        doc_data = doc.to_dict()
        if "ingredients" in doc_data:
            result_map[doc.id] = doc_data["ingredients"]
    
    # 更新游标为当前页最后一条文档
    last_doc = current_page[-1]

注意事项 #

• 如果ingredients数组内存储的是Firestore文档引用类型（DocumentReference），需要对每个引用单独调用.get()方法拉取对应文档内容后再做映射
• 服务账号密钥属于敏感凭证，不要硬编码在代码里、不要提交到公共代码仓库，建议通过环境变量读取
• 文档量级超过10万的场景，不建议直接通过API遍历拉取，优先用Firestore官方导出功能把数据同步到BigQuery/云存储后再做批量处理，效率更高

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