Python上传文件到Google Cloud Storage桶时捕获异常的正确方法

异常类型查找方法 #

你需要分别覆盖两类异常场景，所有具体异常都可以从两个来源枚举：

• 本地文件操作异常：upload_from_filename 首先会读取本地文件，会触发 Python 标准 IO 异常，常见类型包括 FileNotFoundError（本地文件不存在）、PermissionError（本地文件无读权限）、IsADirectoryError（传入的源路径是目录而非文件）
• GCS 交互相关异常：所有 Google Cloud 客户端的服务异常都定义在 google.api_core.exceptions 模块下，和上传场景相关的高频具体异常包括：
  • Unauthenticated：客户端身份认证失败，多为密钥文件无效、过期或环境变量配置错误
  • NotFound：指定的 GCS 桶不存在，或当前账号无权限访问目标桶
  • PermissionDenied：当前账号缺少目标桶的上传权限（缺少 storage.objects.create 权限）
  • BadRequest：请求参数非法，比如目标 blob 名包含 GCS 不允许的特殊字符、内容大小超出限制
  • RequestTimeout：单次上传请求超时，多为网络波动导致
  • ResourceExhausted：GCS 配额不足，比如上传请求 QPS 超过限制、存储容量耗尽
  • RetryError：内置重试次数耗尽仍上传失败，多为服务端临时不可用导致

完整异常捕获实现 #

异常捕获遵循先具体后宽泛的原则，先处理细分场景异常做针对性提示，最后用父类异常兜底避免漏判：

from google.cloud import storage
from google.api_core.exceptions import (
    NotFound, PermissionDenied, Unauthenticated,
    BadRequest, RetryError, RequestTimeout, ResourceExhausted,
    GoogleAPICallError
)
import os

def upload_to_gcp(source_file_name, destination_blob_name):
    # 前置参数校验，提前拦截低级错误，减少无效API请求
    if not os.path.isfile(source_file_name):
        print(f"参数错误：源文件{source_file_name}不存在或不是普通文件")
        return False
    
    try:
        storage_client = storage.Client()
        bucket = storage_client.bucket("some-bucket")
        blob = bucket.blob(destination_blob_name)
        blob.upload_from_filename(source_file_name)
        return True
    # 本地文件异常处理
    except FileNotFoundError:
        print(f"本地文件错误：文件{source_file_name}不存在")
        return False
    except PermissionError:
        print(f"本地文件错误：无权限读取文件{source_file_name}")
        return False
    # 细分GCS异常处理
    except Unauthenticated:
        print("GCS认证失败：请检查密钥配置是否正确")
        return False
    except NotFound:
        print("GCS资源错误：目标桶不存在或无权限访问")
        return False
    except PermissionDenied:
        print("GCS权限错误：当前账号无桶上传权限")
        return False
    except BadRequest as e:
        print(f"GCS请求错误：参数非法，详情：{str(e)}")
        return False
    except RequestTimeout:
        print("上传超时：请检查网络后重试")
        return False
    except ResourceExhausted:
        print("GCS配额不足：请提升配额或稍后重试")
        return False
    except RetryError:
        print("上传失败：多次重试后服务仍不可用")
        return False
    # GCS异常兜底，覆盖所有未枚举的服务端异常
    except GoogleAPICallError as e:
        print(f"GCS服务异常：错误码{e.code}，详情：{str(e)}")
        return False
    # 全局兜底，覆盖所有未知异常
    except Exception as e:
        print(f"未知异常：{str(e)}")
        return False

补充说明 #

如果需要更精细化的错误处理，可以在捕获异常后读取异常对象的 code（HTTP状态码）、details 属性获取更具体的错误信息，自定义对应处理逻辑。

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