如何编写Cloud Function监听并响应GCS Bucket文件增改删事件？

下面是一个极简的Python Cloud Functions示例，可监听指定GCS Bucket的文件创建/更新、删除事件，并输出关键信息到日志：

import functions_framework

# 触发类型：GCS 对象最终化（创建/更新）、删除事件
@functions_framework.cloud_event
def gcs_file_event_listener(cloud_event):
    # 解析事件数据
    data = cloud_event.data
    bucket_name = data["bucket"]
    file_name = data["name"]
    event_type = cloud_event["type"]

    # 根据事件类型处理逻辑
    if event_type == "google.cloud.storage.object.v1.finalized":
        print(f"文件已创建/更新：Bucket={bucket_name}, 文件={file_name}")
        # 这里可以添加你的业务逻辑，比如读取文件内容、同步到其他服务等
    elif event_type == "google.cloud.storage.object.v1.deleted":
        print(f"文件已删除：Bucket={bucket_name}, 文件={file_name}")
        # 这里添加删除后的处理逻辑，比如清理关联数据、发送通知等
    else:
        print(f"未处理的事件类型：{event_type}")

关键说明 #

• 函数通过@functions_framework.cloud_event装饰器声明为Cloud Event触发类型，适配GCS的事件格式
• cloud_event参数包含了事件的所有元数据：
  • cloud_event["type"]：区分事件类型，finalized对应文件创建/更新，deleted对应文件删除
  • data["bucket"]：事件触发的Bucket名称
  • data["name"]：触发事件的文件路径/名称

部署步骤 #

1. 保存上述代码为main.py文件
2. 使用gcloud命令部署（替换占位符为你的实际信息）：

gcloud functions deploy gcs_file_event_listener \
  --runtime python311 \
  --trigger-resource YOUR_BUCKET_NAME \
  --trigger-event google.cloud.storage.object.v1.finalized \
  --trigger-event google.cloud.storage.object.v1.deleted

• --runtime：指定Python版本（选择GCP支持的版本，比如python311）
• --trigger-resource：替换为你要监听的Bucket名称
• --trigger-event：指定要监听的事件类型，这里同时监听了创建/更新和删除事件

验证方法 #

部署完成后，在你的GCS Bucket中上传、修改或删除文件，然后进入Cloud Functions的日志页面，就能看到函数输出的对应事件信息。

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