如何在PyCharm中使用boto3库删除S3存储桶指定扩展名文件

前置准备 #

• 打开PyCharm底部终端，激活当前项目的虚拟环境，执行pip install boto3安装依赖。
• 配置AWS访问凭证：不要把密钥硬编码在脚本里，推荐两种方式：一是在本地配置~/.aws/credentials认证文件，二是在PyCharm对应运行配置的环境变量中添加AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY、AWS_DEFAULT_REGION三个参数，和你账号的AWS信息保持一致。
• 确认使用的凭证拥有目标S3桶的s3:ListBucket、s3:DeleteObject权限，否则会触发权限报错。

实现逻辑说明 #

S3没有提供按后缀直接批量删除的原生接口，整体流程为：通过分页接口拉取桶内全量对象（解决单次列举最多返回1000个对象的限制），筛选出后缀为.csv的非目录对象，每凑够1000个就调用批量删除接口提交请求，减少接口调用开销。

可直接运行的脚本代码 #

import boto3
from botocore.exceptions import ClientError

def _execute_batch_delete(s3_client, bucket_name: str, delete_targets: list) -> None:
    """执行单次批量删除操作，单次最多支持1000个对象"""
    try:
        resp = s3_client.delete_objects(
            Bucket=bucket_name,
            Delete={
                "Objects": delete_targets,
                "Quiet": False
            }
        )
        success_count = len(resp.get("Deleted", []))
        print(f"[+] 本次成功删除 {success_count} 个.csv文件")
        err_list = resp.get("Errors", [])
        if err_list:
            print(f"[!] 本次删除存在 {len(err_list)} 个失败条目：")
            for err in err_list:
                print(f"    - {err['Key']}：错误码{err['Code']}，{err['Message']}")
    except ClientError as e:
        print(f"[!] 批量删除请求失败：{e.response['Error']['Message']}")

def delete_s3_csv_files(bucket_name: str, target_prefix: str = "") -> None:
    """
    删除指定S3桶下所有.csv后缀文件
    :param bucket_name: 目标存储桶名称
    :param target_prefix: 可选，指定要扫描的文件前缀（即指定文件夹路径），不传则扫描全桶
    """
    s3_client = boto3.client("s3")
    paginator = s3_client.get_paginator("list_objects_v2")
    page_iter = paginator.paginate(Bucket=bucket_name, Prefix=target_prefix)

    pending_delete = []
    for page in page_iter:
        if "Contents" not in page:
            print(f"[*] 目标路径下未检索到任何文件")
            return
        for obj in page["Contents"]:
            key = obj["Key"]
            # 跳过S3目录占位符，匹配.csv后缀，默认大小写不敏感
            if not key.endswith("/") and key.lower().endswith(".csv"):
                pending_delete.append({"Key": key})
                # 达到单次批量删除上限就提交请求
                if len(pending_delete) == 1000:
                    _execute_batch_delete(s3_client, bucket_name, pending_delete)
                    pending_delete = []
    # 处理最后一批不足1000个的文件
    if pending_delete:
        _execute_batch_delete(s3_client, bucket_name, pending_delete)
    print("[*] 所有.csv后缀文件处理完成")

if __name__ == "__main__":
    # 替换为实际的存储桶名称
    TARGET_BUCKET = "替换为你的S3桶名"
    # 如果只需要删除某个文件夹下的csv，把路径填到第二个参数即可，比如 "data/2024/"
    delete_s3_csv_files(TARGET_BUCKET)

使用提示 #

• 首次运行建议先把_execute_batch_delete的调用注释掉，先打印pending_delete列表，确认筛选出的文件全是需要删除的.csv文件，避免误删。
• 如果需要严格匹配小写.csv后缀（忽略.CSV等大写后缀文件），把代码中key.lower().endswith(".csv")修改为key.endswith(".csv")即可。
• 如果桶内文件量极大，可以结合S3生命周期规则做批量清理，比脚本遍历效率更高。

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