【Python程序开发系列】对象存储服务AWS S3以及使用示例（案例分析）
这是我的第448篇原创文章。
一、引言
Amazon Simple Storage Service (S3)是 AWS 提供的对象存储服务，具有以下核心特点：
对象存储：存储的是文件对象（而非块存储），每个对象包含数据、元数据和唯一键
无限扩展：可存储任意数量的文件，单个文件最大 5TB
高持久性：承诺 99.999999999%（11个9）的数据持久性
全球访问：通过 HTTP/HTTPS 协议从任何地方访问
成本效益：按实际使用量付费，无需预付费
核心概念：
Bucket（存储桶）：顶级容器，类似文件夹，名称全球唯一
Object（对象）：存储的基本实体，由键（Key）、值（数据）和元数据组成
Key（键）：对象在 Bucket 中的唯一标识，类似文件路径
Value（值）：对象在Bucket这的数据（如文件、字符串、字节等形式）
Meta-Data（元数据）：包括系统指定的文件类型、创建时间、加密算法以及用户上传时指定的元信息，元数据在对象创建后都无法更改。
Tag（标签）：可以为对象指定最多10个标签，标签的键和值最大长度时128和256个字符，相较于元数据，标签是可以修改和新增的，它最大的好处是可以结合权限控制、生命周期管理、和数据分析等使用。
二、实现过程
2.1 安装 SDK
pip install boto3 # AWS 官方 Python SDK2.2 基础配置
方式一：使用 AWS CLI 配置（推荐）
aws configure # 输入 AWS Access Key ID # 输入 AWS Secret Access Key # 输入区域（如 us-east-1）方式二：代码中直接配置
import boto3 # 创建 S3 客户端 s3 = boto3.client( 's3', aws_access_key_id='你的访问密钥', aws_secret_access_key='你的秘密密钥', region_name='us-east-1' )2.3 常用操作示例
创建 S3 客户端或资源对象
import boto3 from botocore.exceptions import ClientError # 创建 S3 客户端或资源对象（资源对象更面向对象） s3_client = boto3.client('s3') s3_resource = boto3.resource('s3')创建 Bucket
def create_bucket(bucket_name): try: s3_client.create_bucket(Bucket=bucket_name) print(f"✅ Bucket '{bucket_name}' 创建成功") except ClientError as e: print(f"❌ 创建失败: {e}")上传文件
def upload_file(file_path, bucket_name, object_name=None): """上传文件到 S3""" if object_name is None: object_name = file_path.split('/')[-1] try: s3_client.upload_file(file_path, bucket_name, object_name) print(f"✅ 文件 '{file_path}' 上传成功") print(f" s3://{bucket_name}/{object_name}") except ClientError as e: print(f"❌ 上传失败: {e}")下载文件
def download_file(bucket_name, object_name, file_path): """从 S3 下载文件""" try: s3_client.download_file(bucket_name, object_name, file_path) print(f"✅ 文件下载成功: {file_path}") except ClientError as e: print(f"❌ 下载失败: {e}")列出 Bucket 中的文件
def list_files(bucket_name): """列出 Bucket 中的所有对象""" try: response = s3_client.list_objects_v2(Bucket=bucket_name) if 'Contents' in response: print(f"\n📁 Bucket '{bucket_name}' 中的文件:") for obj in response['Contents']: print(f" - {obj['Key']} (大小: {obj['Size']} bytes)") else: print(f"Bucket '{bucket_name}' 为空") except ClientError as e: print(f"❌ 列出文件失败: {e}")删除文件
def delete_file(bucket_name, object_name): """删除 S3 中的文件""" try: s3_client.delete_object(Bucket=bucket_name, Key=object_name) print(f"✅ 文件 '{object_name}' 删除成功") except ClientError as e: print(f"❌ 删除失败: {e}")生成预签名 URL（临时访问链接）
def generate_presigned_url(bucket_name, object_name, expiration=3600): """生成临时访问 URL""" try: url = s3_client.generate_presigned_url( 'get_object', Params={'Bucket': bucket_name, 'Key': object_name}, ExpiresIn=expiration ) print(f"🔗 临时访问链接（{expiration}秒后过期）:") print(f" {url}") return url except ClientError as e: print(f"❌ 生成链接失败: {e}") return None上传文件内容（字符串或字节）
def upload_content(bucket_name, object_name, content): """直接上传内容到 S3""" try: s3_client.put_object( Bucket=bucket_name, Key=object_name, Body=content ) print(f"✅ 内容上传成功: s3://{bucket_name}/{object_name}") except ClientError as e: print(f"❌ 上传失败: {e}")2.4 高级功能
使用 S3 Resource（更简洁）
s3 = boto3.resource('s3') bucket = s3.Bucket('my-bucket') # 上传 bucket.upload_file('/tmp/test.txt', 'folder/test.txt') # 下载 bucket.download_file('folder/test.txt', '/tmp/test.txt') # 遍历所有对象 for obj in bucket.objects.all(): print(obj.key, obj.last_modified) # 批量删除 bucket.objects.filter(Prefix='temp/').delete()作者简介：
读研期间发表6篇SCI数据挖掘相关论文，现在某研究院从事数据算法相关科研工作，结合自身科研实践经历不定期分享关于Python、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创，以最简单的方式理解和学习，关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。