如何在Firebase实时数据库路径中批量创建多条记录？

我基于Python开发Web应用，使用Firebase实时数据库存储数据。其中一个接口支持客户端上传CSV文件，将CSV的每一行转换为数据库指定路径下的独立节点（已实现CSV转字典列表的逻辑）。

当前实现的问题是，每行数据都单独调用一次Firebase API，当CSV行数较多时，处理速度极慢：

def create(self, path:str, data: Dict[str, Any]):
        return self.db.reference(path).push(data)

@router.post("/")
def create_alerts(file: UploadFile):
    try:
        data = convert_csv(file) # 将CSV转为字典列表
        for alert in data:
            info_alert = alert
            info_alert["status"] = "not_validated"
            fb_client.create(settings.FB_PATH, info_alert) # 调用上述create方法
    except Exception as e:
        print(e)
        return {"message": "Error creating document"}
    return {"message": "Document created"}

我曾尝试把API调用移出循环，但这样只会在指定路径下创建一个包含所有记录的节点，不符合需求：

try:
        data = convert_csv(file)
        for alert in data:
            info_alert = alert
            info_alert["status"] = "not_validated"
        fb_client.create(settings.FB_PATH, info_alert)

请问有没有办法通过单次API调用，为每个字典在指定路径下创建对应的独立节点？

解决方案 #

Firebase实时数据库支持批量写入，可以通过构造包含所有子节点数据的字典，一次性提交到指定路径。核心思路是：为每条数据生成Firebase风格的唯一push ID，然后将这些ID作为键、对应数据作为值，组成一个大字典，最后通过单次API调用完成写入。

你的更新代码思路是正确的，这里提供优化后的完整实现：

优化后的代码

from firebase_admin import db
from typing import Dict, Any
from fastapi import UploadFile, APIRouter

router = APIRouter()

# 优化create方法，支持批量写入
def create(self, path: str, data: Dict[str, Any]):
        # 使用update方法：保留路径下原有节点，新增批量生成的节点
        return self.db.reference(path).update(data)
        # 若需清空路径后写入新数据，改用set方法：
        # return self.db.reference(path).set(data)

@router.post("/")
def create_alerts(file: UploadFile):
    try:
        data = convert_csv(file) # 将CSV转为字典列表
        ref = fb_client.db.reference(settings.FB_PATH)
        nodes_data = {}
        
        for alert in data:
            info_alert = alert
            info_alert["status"] = "not_validated"
            # 生成Firebase标准的唯一push ID
            new_post_ref = ref.push()
            post_id = new_post_ref.key
            nodes_data[post_id] = info_alert
        
        # 单次API调用完成所有节点写入
        fb_client.create(settings.FB_PATH, nodes_data)
    except Exception as e:
        print(e)
        return {"message": "创建记录失败"}
    return {"message": "所有记录创建完成"}

关键说明 #

1. 批量写入逻辑：通过update方法提交包含多个键值对的字典时，Firebase会一次性处理所有条目，每个键对应一个独立子节点，彻底避免多次API调用的网络开销。
2. Push ID一致性：ref.push()生成的ID和逐行调用push生成的格式完全一致（基于时间戳的唯一值），保证数据结构和原有逻辑对齐。
3. 方法选择：update会保留路径下已有数据，仅新增批量节点；set会清空路径下所有原有数据后写入新内容，可根据业务需求切换。

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