这是我的第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 SDK

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