minio

minio 是 [[MinIO]] 官方提供的 [[Python]] SDK，用于操作 MinIO 对象存储服务。

安装

通过 [[pip]] 安装：

1
pip install minio

在 Python 中创建客户端:

1
from minio import Minio
2

3
# 创建 Minio 客户端
4
client = Minio("play.min.io",
5
    access_key="Q3AM3UQ867SPQQA43P2F",
6
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
7
)

然后就可以调用对应 API 进行操作了。

另外如果是内网的 IP 地址，可能需要加上 secure 参数，并设为 False，不然可能会报错: urllib3.exceptions.MaxRetryError: HTTPSConnectionPool(host=‘10.1.1.11’, port=80): Max retries exceeded with url: / (Caused by SSLError(SSLError(1, ‘[SSL] record layer failure (_ssl.c:1000)‘)))

API

Bucket

list_buckets

列出所有 bucket 信息（可访问）: client.list_buckets()

1
buckets = client.list_buckets() # -> list
2

3
for bucket in buckets:
4
    print(bucket.name, bucket.creation_date)

好像只能获取 name 和 creation_date 两个数据。

对象

list_objects

列出对应 bucket 中所有数据: client.list_objects(bucket_name)

1
objects = client.list_objects("my-bucket") # -> generator
2

3
for obj in objects:
4
    print(obj)

可选参数:

• prefix: 指定前缀（如果前缀是目录名称，需要把 recursive 设为 True）
• recursive: 是否递归列出所有子目录

put_object

上传文件 (数据流): client.put_object(bucket_name, object_name, data, length=len(data))

1
client.put_object("my-bucket", "my-object", data, length=len(data), content_type="application/octet-stream")

参数:

• bucket_name: 桶名称
• object_name: 文件名称
• data: 数据流
• length: 数据流长度

可选参数:

• content_type: 内容类型

fput_object

上传文件 (本地文件): client.fput_object(bucket_name, object_name, file_path)

1
client.fput_object("my-bucket", "my-object", "local/file.txt", content_type="application/octet-stream")

参数:

• bucket_name: 桶名称
• object_name: 文件名称
• file_path: 文件路径

可选参数:

• content_type: 内容类型

get_object

下载文件 (数据流): client.get_object(bucket_name, object_name) -> urllib3.response.HTTPResponse

1
response = client.get_object("my-bucket", "my-object")
2

3
# 保存到本地
4
with open("./image.jpg", "wb") as f:
5
    f.write(response.data)

参数:

• bucket_name: 桶名称
• object_name: 文件名称

fget_object

下载文件 (到本地): client.fget_object(bucket_name, object_name, file_path)

1
client.fget_object("my-bucket", "my-object", "./image.jpg")

参数:

• bucket_name: 桶名称
• object_name: 文件名称
• file_path: 本地保存文件路径

get_presigned_url

获取文件的预签名 URL: client.get_presigned_url(method, bucket_name, object_name)

1
client.get_presigned_url("GET", "my-bucket", "my-object") # -> str

参数:

• method: 请求方法，如 GET、PUT、POST 等
• bucket_name: 桶名称
• object_name: 文件名称

可选参数:

• expires: 过期时间（单位：秒）

使用

按目录删除

1
from minio import Minio
2
from minio.deleteobjects import DeleteObject
3

4
delete_object_list = map(
5
    lambda x: DeleteObject(x.object_name),
6
    client.list_objects("my-bucket", "my/prefix/", recursive=True),
7
)
8
errors = client.remove_objects("my-bucket", delete_object_list)
9
for error in errors:
10
    print("error occured when deleting object", error)

最后循环的时候才会实际执行删除操作。

获取数量

通过 [[#list_objects]] 获取到对象后，计算数量:

1
objects = client.list_objects("my-bucket")
2
sum(1 for _ in objects)

计算后如果需要继续使用 objects，最好是重新再获取一次。

参考

• minio/minio-py
• MinIO | Python Client API Reference
• Python SDK | AIStor Object Store Documentation
• python 操作 Minio