这是我的第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、机器学习、深度学习、人工智能系列基础知识与应用案例。致力于只做原创,以最简单的方式理解和学习,关注我一起交流成长。需要数据集和源码的小伙伴可以关注底部公众号添加作者微信。