如何避免修改GCS中JSON值后Cloud Function（Python）循环执行

问题根源 #

你遇到的循环执行，是因为函数修改并保存my_jsonfile.json回原存储桶时，会再次触发GCS的OBJECT_FINALIZE事件，导致函数反复被调用。

解决方案 #

以下是几种直接可行的修复方式，结合你的现有代码调整：

方案1：通过自定义元数据标记已处理文件 #

在首次处理JSON文件时，给它添加自定义元数据（比如processed: true），函数触发时先检查该标记，仅处理未标记的文件。

修改后的代码示例：

def hello_gcs(event, context):
    bucket_name = event['bucket']
    # 处理文件不在子文件夹的情况，避免split报错
    try:
        folder_name, file_name = event['name'].split('/')
    except ValueError:
        return
    blob_json_file = folder_name + '/my_jsonfile.json'

    if file_name == 'my_jsonfile.json':
        storage_client = storage.Client()
        bucket = storage_client.get_bucket(bucket_name)
        blob = bucket.blob(blob_json_file)
        
        # 检查是否已处理过，避免循环
        if blob.metadata and blob.metadata.get('processed') == 'true':
            print("文件已处理，跳过执行")
            return
        
        json_data, blob = extract_blob_json(bucket_name, blob_json_file)
        has_updates = False
        
        # 先批量处理所有空值字段，再统一保存
        for blob_key, blob_val in json_data.items():
            if blob_val == "":
                gcs_file_name = folder_name + '/' + blob_key + '.csv'
                file_reference_name = blob_key + '.csv'
                object_blob = create_blob(gcs_file_name, bucket_name, file_reference_name)
                json_data[blob_key] = object_blob.id
                print(f"Object name {blob_key} : Corresponding blob ID {object_blob.id} ")
                has_updates = True
        
        if has_updates:
            # 一次性更新JSON内容
            with blob.open("w") as f:
                f.write(json.dumps(json_data))
            # 添加已处理标记
            blob.metadata = {'processed': 'true'}
            blob.patch()

# 以下函数保持原有逻辑不变
def create_blob(gcs_file_name, bucket_name, filename: str) -> Blob:
    # 你的内部工具代码
    return response

def extract_blob_json(bucket_name, file_name):
    storage_client = storage.Client()
    bucket = storage_client.get_bucket(bucket_name)
    blob = bucket.blob(file_name)
    content = blob.download_as_text()
    json_data = json.loads(content)
    return json_data, blob

方案2：跳过自身触发的事件 #

通过判断触发事件的服务账号，跳过由Cloud Function自身发起的文件更新事件。需要先获取你的Cloud Function使用的服务账号（通常格式为[项目ID]@appspot.gserviceaccount.com）：

def hello_gcs(event, context):
    bucket_name = event['bucket']
    try:
        folder_name, file_name = event['name'].split('/')
    except ValueError:
        return
    blob_json_file = folder_name + '/my_jsonfile.json'

    # 跳过函数自身触发的事件
    FUNCTION_SERVICE_ACCOUNT = '你的项目ID@appspot.gserviceaccount.com'
    triggering_account = event.get('metadata', {}).get('service_account')
    if triggering_account == FUNCTION_SERVICE_ACCOUNT:
        print("事件由函数自身触发，跳过")
        return

    if file_name == 'my_jsonfile.json':
        # 原有处理逻辑...

方案3：拆分存储桶（可选） #

如果允许调整存储结构，将原始JSON上传到触发桶，处理后的JSON保存到另一个结果桶，从根源上避免触发循环。此方案适合可以修改上传路径的场景。

关键优化点 #

• 避免多次保存JSON：原代码每次修改一个字段就保存一次，会多次触发事件，建议批量处理所有更新后统一保存。
• 处理路径异常：确保文件不在根目录时函数不会报错，添加split('/')的异常捕获。

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