下面是楼主个人珍藏了好多年的 Python 代码
这段代码用来对兼容 S3 接口的对象存储增删改查，
其实不仅 Cloudflare 的 R2 能用，Backblaze 的 B2 也能用，
甚至是 缤纷云 S4，只需要换一下配置就行了～
import io
import boto3
# 存储桶访问密钥ID，例子：'e025aab23d4a677xxxss999315439bd76e'
S3_ACCESS_KEY_ID = ''
# 存储桶访问密钥，例子：'107a4214ea73bb730bffb59726sxxxx9ca2f3f5bb42ssdsda294d2aa9f3c91bf'
S3_SECRET_ACCESS_KEY = ''
# 存储端点URL，例子：'https://dd645c3sssdd21e57cd7edbbdb4730e445c.r2.cloudflarestorage.com'
S3_ENDPOINT_URL = ''
# 存储桶位置，例子：'WNAM'
S3_REGION_NAME = ''
# 存储桶名称，例子：'mybucket'
S3_BUCKET_NAME = ''
def get_s3_client():
"""
获取S3客户端连接
:return: S3客户端实例
"""
return boto3.client(
's3',
aws_access_key_id=S3_ACCESS_KEY_ID,
aws_secret_access_key=S3_SECRET_ACCESS_KEY,
endpoint_url=S3_ENDPOINT_URL,
region_name=S3_REGION_NAME,
)
def s3_exists(path, bucket_name=S3_BUCKET_NAME):
"""
检查S3中是否存在指定文件
:param path: 文件路径
:param bucket_name: 存储桶名称
:return: 如果文件存在返回True，否则返回False
"""
s3_client = get_s3_client()
try:
s3_client.head_object(Bucket=bucket_name, Key=path)
return True
except Exception:
return False
def s3_upload(path, content, bucket_name=S3_BUCKET_NAME):
"""
上传文件到S3存储
:param path: 目标文件路径
:param content: 文件内容（支持字节流或字符串）
:param bucket_name: 存储桶名称
"""
if isinstance(content, bytes):
file = io.BytesIO(content)
elif isinstance(content, str):
file = io.BytesIO(content.encode())
else:
return
s3_client = get_s3_client()
s3_client.upload_fileobj(file, bucket_name, path)
s3_client.close()
def s3_download(path, bucket_name=S3_BUCKET_NAME):
"""
从S3下载文件
:param path: 文件路径
:param bucket_name: 存储桶名称
:return: 返回的是字节流, 如果需要字符串, 需要使用 decode() 方法
"""
file = io.BytesIO()
s3_client = get_s3_client()
s3_client.download_fileobj(bucket_name, path, file)
s3_client.close()
file.seek(0)
return file.read()
def s3_list(prefix='', bucket_name=S3_BUCKET_NAME):
"""
列出S3中的文件
:param prefix: 文件前缀（用于筛选特定目录下的文件）
:param bucket_name: 存储桶名称
:return: 返回文件信息列表
"""
file_list = []
s3_client = get_s3_client()
for item in s3_client.get_paginator('list_objects_v2').paginate(Bucket=bucket_name, Prefix=prefix):
for content in item.get('Contents', []):
file_list.append(
{
'path': content['Key'],
'size': content['Size'],
'last_modified': content['LastModified'],
}
)
s3_client.close()
return file_list
def s3_delete(path, bucket_name=S3_BUCKET_NAME):
"""
删除S3中的单个文件
:param path: 文件路径
:param bucket_name: 存储桶名称
"""
s3_client = get_s3_client()
s3_client.delete_object(Bucket=bucket_name, Key=path)
s3_client.close()
def s3_delete_all(prefix='', bucket_name=S3_BUCKET_NAME):
"""
批量删除S3中的文件
:param prefix: 文件前缀（用于筛选要删除的文件）
:param bucket_name: 存储桶名称
"""
for item in s3_list(prefix=prefix, bucket_name=bucket_name):
s3_delete(path=item['path'], bucket_name=bucket_name)
if __name__ == '__main__':
# 使用示例：
# 检查文件是否存在
print(s3_exists(path='test/test.txt'))
# 上传文件示例
s3_upload(path='test/test.txt', content='这是测试文件')
# 列出文件
for item in s3_list(prefix='test/'):
print(item)
# 下载文件示例
print(s3_download(path='test/test.txt').decode())
# 删除文件
print(s3_delete(path='test/test.txt'))
# 再次检查文件是否存在
print(s3_exists(path='test/test.txt'))