Multibucket

多对象存储桶SDK，支持UCloud UFile、阿里云 OSS、腾讯云 COS、七牛云 Kodo、通用 S3 等云存储服务。

安装

pip install multibucket

快速开始

from multibucket import UCloudFileClient

# 初始化客户端
ufile = UCloudFileClient(
    public_key="your_public_key",
    private_key="your_private_key",
    bucket="your-bucket",
    region="cn-bj",
    timeout=30  # 全局超时时间（秒，可选）
)

# 上传文件
with open("image.jpg", "rb") as f:
    resp = ufile.upload("photos/image.jpg", f.read(), content_type="image/jpg")
print(resp.status_code)  # 200

# 上传本地文件（自动猜测 Content-Type）
success = ufile.upload_file("photos/image.jpg", "/path/to/image.jpg")
print(success)  # True/False

# 带重试的上传（默认重试 3 次）
success = ufile.upload_with_retry("backup/data.zip", data, retries=3)
print(success)  # True/False

# 获取文件 URL
url = ufile.get_path_url("photos/image.jpg")
print(url)  # https://your-bucket.cn-bj.ufileos.com/photos/image.jpg

# 生成分享链接（私有空间，默认 20 分钟有效）
share_url = ufile.get_shared_url("photos/image.jpg", expire_time=3600)
print(share_url)

# 判断文件是否存在
exists = ufile.is_exists("photos/image.jpg")
print(exists)  # True/False

# 获取文件内容到内存
data = ufile.get_content("photos/image.jpg")  # bytes
text = ufile.get_content("config.json", as_text=True)  # str

# 下载文件到本地
success = ufile.download("photos/image.jpg", "downloaded.jpg")
print(success)  # True/False

# 分片下载（支持进度回调）
success = ufile.multipart_download("videos/large.mp4", "large.mp4")

# 查询文件信息
resp = ufile.head("photos/image.jpg")
print(resp.headers["Content-Length"])

# 列出文件
resp = ufile.list(prefix="photos/", limit=50)
files = resp.json()["DataSet"]

# 删除文件
resp = ufile.delete("photos/image.jpg")
print(resp.status_code)  # 204

# 删除文件（简化版）
success = ufile.delete_file("photos/image.jpg")
print(success)  # True/False

高级功能

自定义域名（CDN）

ufile = UCloudFileClient(
    public_key="your_public_key",
    private_key="your_private_key",
    bucket="your-bucket",
    region="cn-bj",
    domain_url="https://cdn.example.com"  # 自定义 CDN 域名
)

内网访问

ufile = UCloudFileClient(
    public_key="your_public_key",
    private_key="your_private_key",
    bucket="your-bucket",
    region="cn-wlcb",
    internal=True  # 使用内网访问
)

分片上传（大文件）

# 自动分片上传，支持进度回调
def progress_callback(part_number, total_parts, uploaded, total):
    print(f"进度: {part_number + 1}/{total_parts}")

result = ufile.multipart_upload(
    key="videos/large.mp4",
    file_path="/path/to/large.mp4",
    content_type="video/mp4",
    callback=progress_callback
)

# 直接传入二进制内容进行分片上传
result = ufile.multipart_upload_bytes(
    key="matting/2026/08/20/example.webp",
    data=binary_content,
    content_type="image/webp"
)

自定义 HTTP 客户端

可以通过 http_client 参数传入自定义的 httpx.Client 实例：

import httpx
from multibucket import UCloudFileClient

# 创建自定义 client
custom_client = httpx.Client(
    timeout=httpx.Timeout(connect=5.0, read=30.0),
    limits=httpx.Limits(max_connections=100)
)

# 使用自定义 client
ufile = UCloudFileClient(
    public_key="your_public_key",
    private_key="your_private_key",
    bucket="your-bucket",
    region="cn-bj",
    http_client=custom_client
)

注意: 当传入 http_client 时，timeout 参数不会生效，超时由自定义 client 控制。

阿里云 OSS

快速开始

from multibucket import AliyunOSSClient

# 初始化客户端
oss = AliyunOSSClient(
    access_key_id="your_access_key_id",
    access_key_secret="your_access_key_secret",
    bucket="your-bucket",
    region="oss-cn-hangzhou",
    timeout=30  # 全局超时时间（秒，可选）
)

# 上传文件
with open("image.jpg", "rb") as f:
    resp = oss.upload("photos/image.jpg", f, content_type="image/jpeg")
print(resp.status_code)  # 200

# 上传本地文件
success = oss.upload_file("photos/image.jpg", "/path/to/image.jpg")

# 获取文件 URL
url = oss.get_path_url("photos/image.jpg")
print(url)  # https://your-bucket.oss-cn-hangzhou.aliyuncs.com/photos/image.jpg

# 生成分享链接
share_url = oss.get_shared_url("photos/image.jpg", expire_time=3600)

# 下载文件
success = oss.download("photos/image.jpg", "downloaded.jpg")

# 判断文件是否存在
exists = oss.is_exists("photos/image.jpg")

# 删除文件
success = oss.delete_file("photos/image.jpg")

内网访问

oss = AliyunOSSClient(
    access_key_id="your_access_key_id",
    access_key_secret="your_access_key_secret",
    bucket="your-bucket",
    region="oss-cn-hangzhou",
    internal=True  # 使用内网访问
)

分片上传（大文件）

def progress_callback(part_number, total_parts, uploaded, total):
    print(f"进度: {part_number + 1}/{total_parts}")

result = oss.multipart_upload(
    key="videos/large.mp4",
    file_path="/path/to/large.mp4",
    content_type="video/mp4",
    callback=progress_callback
)

自定义 HTTP 客户端

import httpx
from multibucket import AliyunOSSClient

custom_client = httpx.Client(
    timeout=httpx.Timeout(connect=5.0, read=30.0)
)

oss = AliyunOSSClient(
    access_key_id="your_access_key_id",
    access_key_secret="your_access_key_secret",
    bucket="your-bucket",
    region="oss-cn-hangzhou",
    http_client=custom_client
)

注意: 当传入 http_client 时，timeout 参数不会生效，超时由自定义 client 控制。

七牛云 Kodo

快速开始

from multibucket import QiniuKodoClient

# 初始化客户端
qiniu = QiniuKodoClient(
    access_key="your_access_key",
    secret_key="your_secret_key",
    bucket="your-bucket",
    region="z0",             # 区域标识（默认 z0 华东）
    timeout=30               # 全局超时时间（秒，可选）
)

# 上传文件
with open("image.jpg", "rb") as f:
    resp = qiniu.upload("photos/image.jpg", f, content_type="image/jpeg")
print(resp.status_code)  # 200

# 上传本地文件
success = qiniu.upload_file("photos/image.jpg", "/path/to/image.jpg")

# 获取文件 URL
url = qiniu.get_path_url("photos/image.jpg")
print(url)  # http://your-bucket.s3.cn-east-1.qiniucs.com/photos/image.jpg

# 生成分享链接（私有空间）
share_url = qiniu.get_shared_url("photos/image.jpg", expire_time=3600)

# 下载文件
success = qiniu.download("photos/image.jpg", "downloaded.jpg")

# 判断文件是否存在
exists = qiniu.is_exists("photos/image.jpg")

# 删除文件
success = qiniu.delete_file("photos/image.jpg")

# 复制/移动文件
resp = qiniu.copy("photos/image.jpg", "backup/image.jpg")
resp = qiniu.move("photos/old.jpg", "photos/new.jpg")

区域标识

SDK 会根据 region 自动生成 S3 兼容域名，无需手动指定 domain_url。

分片上传（大文件）

def progress_callback(part_number, total_parts, uploaded_bytes, total_bytes):
    pct = uploaded_bytes / total_bytes * 100
    print(f"上传进度: {part_number + 1}/{total_parts} ({pct:.1f}%)")

result = qiniu.multipart_upload(
    key="videos/large.mp4",
    file_path="/path/to/large.mp4",
    content_type="video/mp4",
    callback=progress_callback,
    part_size=4 * 1024 * 1024,  # 4MB 每片（七牛云分片最大 4MB）
)

自定义 HTTP 客户端

import httpx
from multibucket import QiniuKodoClient

custom_client = httpx.Client(
    timeout=httpx.Timeout(connect=5.0, read=30.0)
)

qiniu = QiniuKodoClient(
    access_key="your_access_key",
    secret_key="your_secret_key",
    bucket="your-bucket",
    region="z0",
    http_client=custom_client
)

注意: 当传入 http_client 时，timeout 参数不会生效，超时由自定义 client 控制。

通用 S3 兼容存储

支持 AWS S3、MinIO、Cloudflare R2、腾讯 COS 等任何 S3 兼容服务。

快速开始

from multibucket import S3Client

# AWS S3
client = S3Client(
    access_key="your_access_key",
    secret_key="your_secret_key",
    bucket="your-bucket",
    region="us-east-1",
    timeout=30
)

# MinIO
client = S3Client(
    access_key="your_access_key",
    secret_key="your_secret_key",
    bucket="your-bucket",
    region="us-east-1",
    endpoint="http://localhost:9000"
)

# Cloudflare R2
client = S3Client(
    access_key="your_access_key",
    secret_key="your_secret_key",
    bucket="your-bucket",
    region="auto",
    endpoint="https://<account_id>.r2.cloudflarestorage.com"
)

# 上传文件
with open("image.jpg", "rb") as f:
    resp = client.upload("photos/image.jpg", f, content_type="image/jpeg")
print(resp.status_code)  # 200

# 上传本地文件
success = client.upload_file("photos/image.jpg", "/path/to/image.jpg")

# 获取文件 URL
url = client.get_path_url("photos/image.jpg")

# 生成预签名分享链接
share_url = client.get_shared_url("photos/image.jpg", expire_time=3600)

# 下载文件
success = client.download("photos/image.jpg", "downloaded.jpg")

# 判断文件是否存在
exists = client.is_exists("photos/image.jpg")

# 删除文件
success = client.delete_file("photos/image.jpg")

分片上传（大文件）

def progress_callback(part_number, total_parts, uploaded_bytes, total_bytes):
    pct = uploaded_bytes / total_bytes * 100
    print(f"上传进度: {part_number + 1}/{total_parts} ({pct:.1f}%)")

result = client.multipart_upload(
    key="videos/large.mp4",
    file_path="/path/to/large.mp4",
    content_type="video/mp4",
    callback=progress_callback,
    part_size=5 * 1024 * 1024,  # 5MB 每片（S3 最小分片 5MB）
)

自定义 HTTP 客户端

import httpx
from multibucket import S3Client

custom_client = httpx.Client(
    timeout=httpx.Timeout(connect=5.0, read=30.0)
)

client = S3Client(
    access_key="your_access_key",
    secret_key="your_secret_key",
    bucket="your-bucket",
    region="us-east-1",
    http_client=custom_client
)

注意: 当传入 http_client 时，timeout 参数不会生效，超时由自定义 client 控制。