本文目录导读：
我来详细介绍一下Python操作对象存储的几种主要方式：
使用boto3操作AWS S3（最常用）
安装
pip install boto3
基本操作示例
import boto3
from botocore.exceptions import ClientError
# 创建S3客户端
s3_client = boto3.client(
's3',
aws_access_key_id='YOUR_ACCESS_KEY',
aws_secret_access_key='YOUR_SECRET_KEY',
region_name='us-east-1'
)
# 创建存储桶
def create_bucket(bucket_name):
try:
s3_client.create_bucket(Bucket=bucket_name)
print(f"存储桶 {bucket_name} 创建成功")
except ClientError as e:
print(f"创建失败: {e}")
# 上传文件
def upload_file(file_name, bucket, object_name=None):
if object_name is None:
object_name = file_name
try:
s3_client.upload_file(file_name, bucket, object_name)
print(f"文件 {file_name} 上传成功")
except ClientError as e:
print(f"上传失败: {e}")
# 下载文件
def download_file(bucket, object_name, file_name):
try:
s3_client.download_file(bucket, object_name, file_name)
print(f"文件下载成功")
except ClientError as e:
print(f"下载失败: {e}")
# 列出存储桶中的对象
def list_objects(bucket):
try:
response = s3_client.list_objects_v2(Bucket=bucket)
for obj in response.get('Contents', []):
print(f" - {obj['Key']} (大小: {obj['Size']} bytes)")
except ClientError as e:
print(f"列出失败: {e}")
# 删除对象
def delete_object(bucket, object_name):
try:
s3_client.delete_object(Bucket=bucket, Key=object_name)
print(f"对象 {object_name} 删除成功")
except ClientError as e:
print(f"删除失败: {e}")
# 使用示例
if __name__ == "__main__":
bucket_name = "my-test-bucket"
create_bucket(bucket_name)
upload_file("local_file.txt", bucket_name, "remote_file.txt")
list_objects(bucket_name)
download_file(bucket_name, "remote_file.txt", "downloaded_file.txt")
使用minio库（通用对象存储）
安装
pip install minio
兼容多种对象存储服务
from minio import Minio
from minio.error import S3Error
# 创建客户端（兼容阿里云OSS、腾讯云COS等）
def create_minio_client():
client = Minio(
"play.min.io", # 端点地址
access_key="YOUR_ACCESS_KEY",
secret_key="YOUR_SECRET_KEY",
secure=True # 使用HTTPS
)
return client
# 文件上传
def upload_with_minio(client, bucket_name, file_path, object_name):
try:
# 检查存储桶是否存在
if not client.bucket_exists(bucket_name):
client.make_bucket(bucket_name)
print(f"创建存储桶 {bucket_name}")
# 上传文件
result = client.fput_object(
bucket_name, object_name, file_path
)
print(f"上传成功: {result.object_name}")
except S3Error as e:
print(f"上传失败: {e}")
# 流式上传（适用于大文件）
def upload_large_file(client, bucket_name, file_path, object_name):
try:
with open(file_path, 'rb') as file_data:
# 获取文件大小
file_stat = os.stat(file_path)
result = client.put_object(
bucket_name,
object_name,
file_data,
file_stat.st_size,
content_type='application/octet-stream'
)
print(f"大文件上传成功: {result.object_name}")
except S3Error as e:
print(f"上传失败: {e}")
# 生成临时下载链接
def generate_presigned_url(client, bucket_name, object_name, expiry=3600):
try:
url = client.presigned_get_object(
bucket_name, object_name, expires=expiry
)
print(f"临时下载链接: {url}")
return url
except S3Error as e:
print(f"生成链接失败: {e}")
# 使用示例
client = create_minio_client()
upload_with_minio(client, "my-bucket", "local.txt", "remote.txt")
使用oss2操作阿里云OSS
安装
pip install oss2
示例代码
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
# 认证方式1：使用AK
auth = oss2.Auth('YOUR_ACCESS_KEY', 'YOUR_SECRET_KEY')
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'bucket-name')
# 认证方式2：使用环境变量（推荐）
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'bucket-name')
# 上传文件
def upload_to_oss(local_file, remote_file):
try:
# 简单上传
bucket.put_object_from_file(remote_file, local_file)
print(f"文件上传成功: {remote_file}")
# 流式上传
with open(local_file, 'rb') as f:
bucket.put_object(remote_file, f)
except oss2.exceptions.OssError as e:
print(f"上传失败: {e}")
# 分片上传（大文件）
def multipart_upload(local_file, remote_file):
try:
# 初始化分片上传
upload_id = bucket.init_multipart_upload(remote_file).upload_id
parts = []
# 分片大小（10MB）
part_size = 10 * 1024 * 1024
with open(local_file, 'rb') as f:
part_number = 1
while True:
data = f.read(part_size)
if not data:
break
# 上传分片
result = bucket.upload_part(
remote_file, upload_id, part_number, data
)
parts.append(oss2.models.PartInfo(part_number, result.etag))
part_number += 1
# 完成分片上传
bucket.complete_multipart_upload(remote_file, upload_id, parts)
print("分片上传完成")
except oss2.exceptions.OssError as e:
print(f"分片上传失败: {e}")
# 下载文件
def download_from_oss(remote_file, local_file):
try:
bucket.get_object_to_file(remote_file, local_file)
print(f"文件下载成功: {local_file}")
except oss2.exceptions.OssError as e:
print(f"下载失败: {e}")
使用cos-python-sdk-v5操作腾讯云COS
安装
pip install cos-python-sdk-v5
示例代码
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import sys
import logging
# 配置
secret_id = 'YOUR_SECRET_ID'
secret_key = 'YOUR_SECRET_KEY'
region = 'ap-guangzhou'
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)
client = CosS3Client(config)
# 上传文件
def upload_to_cos(local_file, remote_file, bucket_name):
try:
response = client.upload_file(
Bucket=bucket_name,
LocalFilePath=local_file,
Key=remote_file,
PartSize=10, # 分片大小，单位MB
MAXThread=5 # 并发线程数
)
print(f"文件上传成功: {response['ETag']}")
except Exception as e:
print(f"上传失败: {e}")
# 下载文件
def download_from_cos(remote_file, local_file, bucket_name):
try:
response = client.download_file(
Bucket=bucket_name,
Key=remote_file,
DestFilePath=local_file
)
print(f"文件下载成功")
except Exception as e:
print(f"下载失败: {e}")
通用工具类封装
import os
import logging
from typing import Optional, List
class ObjectStorageClient:
"""对象存储客户端封装"""
def __init__(self, provider='s3', **kwargs):
self.provider = provider
if provider == 's3':
import boto3
self.client = boto3.client('s3', **kwargs)
elif provider == 'minio':
from minio import Minio
self.client = Minio(**kwargs)
elif provider == 'oss':
import oss2
auth = oss2.Auth(kwargs['access_key'], kwargs['secret_key'])
self.client = oss2.Bucket(
auth,
kwargs['endpoint'],
kwargs['bucket_name']
)
else:
raise ValueError(f"不支持的存储类型: {provider}")
def upload(self, local_path: str, remote_path: str,
bucket: Optional[str] = None) -> bool:
"""上传文件"""
try:
if self.provider == 's3':
self.client.upload_file(local_path, bucket, remote_path)
elif self.provider == 'minio':
self.client.fput_object(bucket, remote_path, local_path)
elif self.provider == 'oss':
self.client.put_object_from_file(remote_path, local_path)
return True
except Exception as e:
logging.error(f"上传失败: {e}")
return False
def download(self, remote_path: str, local_path: str,
bucket: Optional[str] = None) -> bool:
"""下载文件"""
try:
if self.provider == 's3':
self.client.download_file(bucket, remote_path, local_path)
elif self.provider == 'minio':
self.client.fget_object(bucket, remote_path, local_path)
elif self.provider == 'oss':
self.client.get_object_to_file(remote_path, local_path)
return True
except Exception as e:
logging.error(f"下载失败: {e}")
return False
def delete(self, remote_path: str, bucket: Optional[str] = None) -> bool:
"""删除对象"""
try:
if self.provider == 's3':
self.client.delete_object(Bucket=bucket, Key=remote_path)
elif self.provider == 'minio':
self.client.remove_object(bucket, remote_path)
elif self.provider == 'oss':
self.client.delete_object(remote_path)
return True
except Exception as e:
logging.error(f"删除失败: {e}")
return False
# 使用示例
storage = ObjectStorageClient(
provider='s3',
aws_access_key_id='YOUR_KEY',
aws_secret_access_key='YOUR_SECRET'
)
storage.upload('local.txt', 'remote.txt', bucket='my-bucket')
注意事项
- 安全性：不要在代码中硬编码密钥，使用环境变量或密钥管理服务
- 大文件处理：使用分片上传/断点续传
- 错误处理：完善异常捕获和重试机制
- 性能优化：使用连接池、并发上传等
- 资源清理：及时关闭连接，释放资源
选择哪个库主要取决于你使用的具体云服务商,boto3是最通用的选择。
