如何使用Firebase Admin SDK实现Python脚本持续扫描Firestore新增文档并获取ID

核心思路 #

不推荐使用定时轮询扫描的方式实现，会浪费Firestore读写配额、延迟高。直接用Firebase Admin SDK内置的实时快照监听器即可，新增文档写入时会主动触发回调，是官方推荐的低消耗、低延迟实现方案。

前置依赖 #

• 安装Firebase Admin SDK：pip install firebase-admin
• 已提前下载Firebase项目的服务账号密钥JSON文件

具体实现代码 #

1. 初始化Firebase Admin SDK #

import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore

# 替换为你的服务账号密钥文件实际路径
cred = credentials.Certificate("serviceAccountKey.json")
firebase_admin.initialize_app(cred)

db = firestore.client()

2. 编写新增文档处理回调 #

你已开发完成的信息处理逻辑可以直接放入该回调中，代码默认只处理新增文档，可按需自行增加修改、删除文档的处理逻辑。

def on_new_doc_added(col_snapshot, changes, read_time):
    for change in changes:
        # 仅过滤新增的文档事件
        if change.type.name == 'ADDED':
            doc_data = change.document.to_dict()
            doc_id = change.document.id
            print(f"捕获新增文档，ID：{doc_id}，内容：{doc_data}")
            # 这里插入你已完成的信息处理代码
            # your_process_logic(doc_data)

3. 启动持续监听 #

替换为你实际需要监听的Firestore集合名称，监听启动后脚本会保持运行，有新文档写入时自动触发回调。

# 替换为你的目标集合名称
target_collection = db.collection("user_input_data")

# 绑定监听器
watch = target_collection.on_snapshot(on_new_doc_added)

# 阻塞进程保持监听运行
while True:
    pass

优化建议 #

• 可在集合引用后增加where条件过滤监听范围，进一步减少不必要的触发：target_collection = db.collection("user_input_data").where("process_status", "==", "unhandled")
• 生产环境部署建议增加异常捕获和自动重连逻辑，避免网络波动导致监听中断
• 服务账号密钥不要硬编码在代码中，生产环境建议通过系统环境变量加载配置

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