如何解码Cloud Run函数中Firestore事件的Protobuf负载

嗨，我看你已经在着手处理Firestore触发的Cloud Run函数里的Protobuf负载了，这就帮你把整个解码流程梳理清楚，确保你能顺利拿到文档的新旧数据~

首先得确保你安装了正确的依赖包，这些是处理Cloud Event和Firestore Protobuf负载必需的：

pip install functions-framework cloudevents google-cloud-events protobuf

接下来咱们一步步拆解你的代码，把每个环节的作用说清楚：

1. 导入必要模块 #

这些模块分别负责Cloud Event处理、Firestore事件结构解析，以及Protobuf到字典的转换：

from cloudevents.http import CloudEvent
import functions_framework
from google.events.cloud import firestore
from google.protobuf.json_format import MessageToDict

2. 定义Cloud Event触发的函数 #

用@functions_framework.cloud_event装饰器标记函数，告诉框架这是一个由Cloud Event触发的函数：

@functions_framework.cloud_event
def hello_firestore(cloud_event: CloudEvent) -> None:
    """Triggers by a change to a Firestore document.
    Args:
        cloud_event: cloud event with information on the firestore event trigger
    """

3. 解析Protobuf负载 #

Firestore事件的负载是二进制的Protobuf格式，咱们需要先把它解析成对应的Python对象：

firestore_payload = firestore.DocumentEventData()
    firestore_payload._pb.ParseFromString(cloud_event.data)

4. 转换为Python字典（方便处理） #

Protobuf对象直接操作不太方便，用MessageToDict把它转成普通的Python字典，同时要处理可能为空的情况（比如创建文档时没有old_value，删除文档时没有new_value）：

old_value_dict = MessageToDict(firestore_payload.old_value._pb) if firestore_payload.old_value else {}
    new_value_dict = MessageToDict(firestore_payload.value._pb) if firestore_payload.value else {}

5. 验证和使用数据 #

最后你就可以打印或者使用这些字典数据了，比如还能顺便拿到触发事件的文档路径：

print(f"触发事件的文档路径: {cloud_event['subject']}")
    print("旧文档数据:", old_value_dict)
    print("新文档数据:", new_value_dict)

如果遇到解析失败的情况，先检查一下你的函数触发的事件类型是不是google.cloud.firestore.document.v1.written，只有这个类型的事件才会用对应的Firestore Protobuf结构哦~

备注：内容来源于stack exchange，提问作者Tokyo