如何检测GCP Bucket中已删除的Metadata文件夹并终止任务？

要检测GCS中Metadata文件夹的存在性，需利用GCS的前缀特性（GCS本身无真实文件夹，均通过Blob路径前缀模拟），在读取文件前完成校验：

方法1：校验前缀下是否存在任何Blob（含文件夹占位或文件） #

修改代码，在遍历文件前先验证目标前缀是否有对应内容：

if len(input_dir) > 0:
    logger.info(f'Fetching GCS data for prefix:{input_dir}')
    gcs_prefix = input_dir.split(f'{bucket.name}/')[-1]
    metadata_prefix = f"{gcs_prefix}Metadata/"
    csv_df = pd.DataFrame()
    
    # 检测Metadata文件夹是否存在
    # 用delimiter='/'筛选前缀匹配的目录或文件
    blobs = storage_client.list_blobs(bucket.name, prefix=metadata_prefix, delimiter='/')
    has_metadata_folder = False
    
    # 检查是否有前缀匹配的文件Blob
    if next(blobs, None) is not None:
        has_metadata_folder = True
    # 检查是否有前缀对应的目录占位
    blobs = storage_client.list_blobs(bucket.name, prefix=metadata_prefix, delimiter='/')
    if hasattr(blobs, 'prefixes') and metadata_prefix in blobs.prefixes:
        has_metadata_folder = True
    
    if not has_metadata_folder:
        logger.error("### Job stopped: Metadata folder does not exist!")
        return empty_extract_df
    
    # 读取CSV文件逻辑
    metadata_files = []
    for blob in storage_client.list_blobs(bucket.name, prefix=metadata_prefix):
        if blob.name.rsplit('.', 1)[-1].lower().strip() in EXTN_CSV:
            metadata_files.append(f"gs://{BUCKET_NAME}/{blob.name}")
    
    if not metadata_files:
        logger.error("### Job stopped: No CSV files found in Metadata folder!")
        return empty_extract_df
    
    meta_files_list = (pd.read_csv(file) for file in metadata_files)
    meta_files_df = pd.concat(meta_files_list)

方法2：检查文件夹占位Blob（若工作流中通过空Blob标记文件夹） #

如果你的流程是通过上传空Blob（如Metadata/）标记文件夹存在，可直接检查该Blob是否存在：

if len(input_dir) > 0:
    logger.info(f'Fetching GCS data for prefix:{input_dir}')
    gcs_prefix = input_dir.split(f'{bucket.name}/')[-1]
    metadata_prefix = f"{gcs_prefix}Metadata/"
    csv_df = pd.DataFrame()
    
    # 检查文件夹占位Blob是否存在
    folder_blob = bucket.blob(metadata_prefix)
    if not folder_blob.exists():
        logger.error("### Job stopped: Metadata folder does not exist!")
        return empty_extract_df
    
    # 读取CSV文件逻辑
    metadata_files = []
    for blob in storage_client.list_blobs(bucket.name, prefix=metadata_prefix):
        if blob.name.rsplit('.', 1)[-1].lower().strip() in EXTN_CSV:
            metadata_files.append(f"gs://{BUCKET_NAME}/{blob.name}")
    
    if not metadata_files:
        logger.error("### Job stopped: No CSV files found in Metadata folder!")
        return empty_extract_df
    
    meta_files_list = (pd.read_csv(file) for file in metadata_files)
    meta_files_df = pd.concat(meta_files_list)

额外注意 #

• 你原代码中if not meta_files_list:的判断无效，因为生成器对象永远不会被判定为“空”，应改为判断metadata_files列表是否为空，这也是之前触发ValueError的原因之一。
• GCS虚拟文件夹会在最后一个文件被删除时自动消失，因此通过前缀检测是最可靠的验证方式。

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