Python脚本如何操作对象存储

本文目录导读：

1. 使用boto3操作AWS S3（最常用）
2. 使用minio库（通用对象存储）
3. 使用oss2操作阿里云OSS
4. 使用cos-python-sdk-v5操作腾讯云COS
5. 通用工具类封装
6. 注意事项

我来详细介绍一下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')

注意事项

1. 安全性：不要在代码中硬编码密钥，使用环境变量或密钥管理服务
2. 大文件处理：使用分片上传/断点续传
3. 错误处理：完善异常捕获和重试机制
4. 性能优化：使用连接池、并发上传等
5. 资源清理：及时关闭连接，释放资源

选择哪个库主要取决于你使用的具体云服务商,boto3是最通用的选择。