4. Python 异常体系基础：内置异常类、自定义异常、异常链与上下文管理器

在 S3 存储接口开发中，网络抖动、权限过期、桶不存在、对象被锁定等异常场景层出不穷。Python 的异常体系为这些错误提供了结构化的处理手段。本章从内置异常类出发，逐步深入到自定义异常、异常链和上下文管理器，为后续构建 S3 错误处理框架打下坚实基础。

4.1 内置异常类层次结构

Python 所有异常都继承自 BaseException，而绝大多数应用级异常继承自 Exception。S3 开发中高频接触的内置异常集中在 OSError、ValueError、TypeError 及其子类。

4.2 自定义异常：为 S3 错误赋予语义

内置异常无法表达 S3 特有的错误语义（如 BucketAlreadyExists、NoSuchKey）。通过继承 Exception 创建自定义异常，可以让错误处理代码更具可读性和可维护性。

class S3Error(Exception):
    """S3 操作基础异常"""
    def __init__(self, message: str, status_code: int = None, request_id: str = None):
        self.status_code = status_code
        self.request_id = request_id
        super().__init__(message)

class BucketAlreadyExists(S3Error):
    """桶已存在（创建桶时冲突）"""
    pass

class NoSuchKey(S3Error):
    """对象键不存在"""
    pass

class AccessDenied(S3Error):
    """访问被拒绝（权限不足）"""
    pass

class BucketNotEmpty(S3Error):
    """桶非空（删除桶时违反约束）"""
    pass

4.3 异常链：raise ... from 保留错误上下文

在 S3 客户端封装中，经常需要将底层网络异常转换为业务异常。直接 raise 会丢失原始异常栈，而 raise ... from 可以保留完整的异常链。

def get_object(bucket: str, key: str) -> bytes:
    try:
        response = s3_client.get_object(Bucket=bucket, Key=key)
        return response['Body'].read()
    except ConnectionError as e:
        # 保留原始异常上下文，抛出业务异常
        raise S3Error(
            message=f"无法连接到 S3 端点，获取对象 {key} 失败",
            status_code=503
        ) from e
    except ClientError as e:
        if e.response['Error']['Code'] == 'NoSuchKey':
            raise NoSuchKey(f"对象 {key} 在桶 {bucket} 中不存在") from e
        raise S3Error(
            message=f"S3 客户端错误: {e.response['Error']['Message']}",
            status_code=e.response['ResponseMetadata']['HTTPStatusCode']
        ) from e

4.4 上下文管理器与异常

S3 操作中大量涉及资源管理：HTTP 连接、文件句柄、临时凭证。上下文管理器（with 语句）确保无论是否发生异常，资源都能被正确释放。

4.4.1 使用 contextmanager 装饰器简化 S3 资源管理

from contextlib import contextmanager

@contextmanager
def s3_client_context(endpoint_url: str, access_key: str, secret_key: str):
    """安全的 S3 客户端上下文管理器，自动关闭连接"""
    client = boto3.client(
        's3',
        endpoint_url=endpoint_url,
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key
    )
    try:
        yield client
    finally:
        # 关闭底层 HTTP 连接池
        client.close()
        print("[S3Client] 连接已释放")

# 使用示例
with s3_client_context("https://s3.example.com", "ak", "sk") as s3:
    data = s3.get_object(Bucket="my-bucket", Key="file.txt")
    # 即使 get_object 抛出异常，client.close() 也会执行

4.4.2 上下文管理器中的异常传播规则

4.5 综合示例：S3 对象下载的错误处理

将内置异常、自定义异常、异常链和上下文管理器整合到一个完整的 S3 对象下载函数中。

class S3DownloadError(S3Error):
    """S3 下载专用异常"""
    pass

@contextmanager
def s3_download_session(bucket: str, key: str):
    """S3 下载会话上下文，自动处理连接生命周期"""
    client = boto3.client('s3')
    try:
        yield client
    except (ConnectionError, TimeoutError) as e:
        raise S3DownloadError(
            message=f"下载 {bucket}/{key} 时网络异常",
            status_code=503,
            request_id=getattr(e, 'request_id', None)
        ) from e
    except ClientError as e:
        code = e.response['Error']['Code']
        if code == 'NoSuchKey':
            raise NoSuchKey(f"对象 {key} 不存在") from e
        elif code == 'AccessDenied':
            raise AccessDenied(f"无权访问 {bucket}/{key}") from e
        else:
            raise S3DownloadError(
                message=f"S3 服务端错误: {e.response['Error']['Message']}",
                status_code=e.response['ResponseMetadata']['HTTPStatusCode']
            ) from e
    finally:
        client.close()

def download_s3_object(bucket: str, key: str) -> bytes:
    """安全下载 S3 对象，返回 bytes 内容"""
    with s3_download_session(bucket, key) as s3:
        response = s3.get_object(Bucket=bucket, Key=key)
        return response['Body'].read()