Python 从 Cloud Firestore 查询数据触发 KeyError 报错如何解决

• 核心问题是Cloud Firestore为无模式文档数据库，同一集合内的不同文档不需要具备完全相同的字段，你的代码直接用遍历到的第一个文档的字段集合作为全局表头，后续遍历到缺少Media字段的文档时，强行读取row_items[key]就会触发KeyError。
• 代码还存在两个隐含问题：
  • 每写入一个字段就执行一次out_file.write("\n")，会导致单条文档的所有字段被拆分到不同行，CSV结构完全错误
  • 最后直接读取doc.to_dict()['StartDate']，如果文档缺少该字段也会触发KeyError

推荐优先使用pandas实现导出逻辑，自动处理缺失值、格式转义等问题：

import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
import pandas as pd

# 初始化Firebase
cred = credentials.Certificate('creddatabase.json')
firebase_admin.initialize_app(cred)
db = firestore.client()

# 先拉取所有文档，同时收集全量字段
all_docs = []
all_keys = set()
for doc in db.collection(u'testcollection').stream():
    doc_dict = doc.to_dict()
    all_docs.append(doc_dict)
    all_keys.update(doc_dict.keys())

# 对表头排序
sorted_keys = sorted(list(all_keys))

# 生成DataFrame自动补全缺失字段
df = pd.DataFrame(all_docs, columns=sorted_keys)
# 格式化日期字段，不存在则跳过不会报错
if 'StartDate' in df.columns:
    df['StartDate'] = df['StartDate'].dt.strftime('%Y/%m/%d')
# 导出CSV，自动处理逗号转义
df.to_csv('testcollection.csv', index=False, encoding='utf-8-sig')

如果要保留原生写法不依赖pandas，修改原有逻辑即可：

import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore

# 初始化Firebase
cred = credentials.Certificate('creddatabase.json')
firebase_admin.initialize_app(cred)
db = firestore.client()
users_ref = db.collection(u'testcollection')

# 先收集所有文档和全量字段
all_docs = []
all_keys = set()
for doc in users_ref.stream():
    row = doc.to_dict()
    all_docs.append(row)
    all_keys.update(row.keys())
rows_keys = sorted(list(all_keys))

# 写入CSV
out_file = open("testcollection.csv","w", encoding='utf-8')
# 写表头
out_file.write(','.join([k.replace(',','_') for k in rows_keys]) + '\n')
# 写内容
for row in all_docs:
    row_values = []
    for key in rows_keys:
        # 用get方法读取，缺失字段返回空字符串避免KeyError
        val = row.get(key, '')
        # 单独处理日期格式
        if key == 'StartDate' and val:
            val = f"{val.year}/{val.month}/{val.day}"
        row_values.append(str(val).replace(',',' '))
    # 整行字段拼接完成后再换行
    out_file.write(','.join(row_values) + '\n')
out_file.close()

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