1、 pip安装minio
pip3 install minio2、文件目录
ImportError: cannot import name 'Minio'
pip3 install minio 还是报错ImportError错误。
原因是建的文件夹名与pytnon文件minio包重名了，qaq。。。
3、连接Minio文件存储服务器
#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
@author:ZSW
@file:
@time:2020/12/04
"""
from minio import Minio
# 使用endpoint、access key和secret key来初始化minioClient对象。
minioClient = Minio('192.168.18.26:9000',
access_key='minioadmin',
secret_key='minioadmin',
secure=False)4、Python操作存储桶(文件夹)
#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
@author:ZSW
@file:
@time:2020/12/04
"""
from minio_learn.MinioConfig import minioClient
from import ResponseError
class Bucket:
# 创建桶(调用make_bucket来创建一个存储桶) make_bucket函数
"""
注：创建桶命名限制：小写字母，句点，连字符和数字是
唯一允许使用的字符（使用大写字母、下划线等命名会报错），长度至少应为3个字符
"""
def create_bucket(self):
try:
if minioClient.bucket_exists(bucket_name='pictures'): # bucket_exists：检查桶是否存在
print("该存储桶已经存在")
else:
minioClient.make_bucket("pictures")
print("存储桶创建成功")
except ResponseError as err:
print(err)
# 列出所有的存储桶 list_buckets函数
def get_bucket_list(self):
try:
buckets = minioClient.list_buckets()
for bucket in buckets:
print(, bucket.creation_date) # 获取桶的名称和创建时间
except ResponseError as err:
print(err)
# 删除存储桶
def get_remove_bucket(self):
try:
minioClient.remove_bucket("pictures")
print("删除存储桶成功")
except ResponseError as err:
print(err)
# 列出存储桶中所有对象 或者使用 list_objects_v2也可
def get_bucket_files(self):
try:
objects = minioClient.list_objects('testfiles', prefix=None,
recursive=True) # prefix用于过滤的对象名称前缀
for obj in objects:
print(obj.bucket_name, obj.object_name.encode('utf-8'), obj.last_modified,
, , obj.content_type)
except ResponseError as err:
print(err)
# 列出存储桶中未完整上传的对象
def get_list_incomplete_uploads(self):
try:
uploads = minioClient.list_incomplete_uploads('testfiles',
prefix=None,
recursive=True)
for obj in uploads:
print(obj.bucket_name, obj.object_name, obj.upload_id, )
except ResponseError as err:
print(err)
# 获取存储桶的当前策略
def bucket_policy(self):
try:
policy = minioClient.get_bucket_policy('testfiles')
print(policy)
except ResponseError as err:
print(err)
# # 给指定的存储桶设置存储桶策略
# def get_set_bucket_policy(self):
# try:
# minioClient.set_bucket_policy('testfiles', policy.READ_ONLY)
# except ResponseError as err:
# print(err)
# 获取存储桶上的通知配置
def bucket_notification(self):
try:
# 获取存储桶的通知配置。
notification = minioClient.get_bucket_notification('testfiles')
print(notification)
# 如果存储桶上没有任何通知：
# notification == {}
except ResponseError as err:
print(err)
# 给存储桶设置通知配置
def set_bucket_notification(bucket_name, notification):
pass
# 删除存储桶上配置的所有通知
def remove_all_bucket_notifications(bucket_name):
try:
minioClient.remove_all_bucket_notifications('mybucket')
except ResponseError as err:
print(err)
# 监听存储桶上的通知
def listen_bucket_notification(bucket_name, prefix, suffix, events):
pass
if __name__ == '__main__':
Bucket().bucket_notification()5、Python操作文件对象
#!/usr/bin/python
# -*- coding: UTF-8 -*-
"""
@author:ZSW
@file:
@time:2020/12/04
"""
from minio_learn.MinioConfig import minioClient
from import ResponseError
import os
class Object:
# 从桶中下载一个对象txt、csv文件都可以
def load_object(self):
try:
data = minioClient.get_object('testfiles', 'long_lat.csv')
with open('./load_files/long_lat.csv', 'wb') as file_data:
for d in (32 * 1024):
file_data.write(d)
print("Sussess")
except ResponseError as err:
print(err)
# 下载一个对象的指定区间的字节数组
def load_partial_object(self):
try:
data = minioClient.get_partial_object('testfiles', '', 2, 8)
with open('./load_files/123_new.txt', 'wb') as file_data:
for d in data:
file_data.write(d)
print("Sussess") # 部分出现乱码
except ResponseError as err:
print(err)
# 下载并将文件保存到本地
def fget_object(self):
try:
print(minioClient.fget_object('testfiles', '', './load_files/123_new1.txt'))
except ResponseError as err:
print(err)
# 拷贝对象存储服务上的源对象到一个新对象
# 注：该API支持的最大文件大小是5GB
# 可通过copy_conditions参数设置copy条件
# 经测试copy复制28M的文件需要663ms; 1.8G的压缩包需要53s
def get_copy_object(self):
try:
copy_result = minioClient.copy_object("pictures", "",
"/testfiles/"
)
print(copy_result)
except ResponseError as err:
print(err)
# 添加一个新的对象到对象存储服务
"""
单个对象的最大大小限制在5TB。put_object在对象大于5MiB时，自动使用multiple parts方式上传。
这样，当上传失败时，客户端只需要上传未成功的部分即可（类似断点上传）。
上传的对象使用MD5SUM签名进行完整性验证。
"""
def upload_object(self):
# 放置一个具有默认内容类型的文件，成功后将打印服务器计算出的etag标识符
try:
with open('./picture_files/', 'rb') as file_data:
file_stat = ('./picture_files/')
print(minioClient.put_object('pictures', '',
file_data, file_stat.st_size))
print("Sussess")
except ResponseError as err:
print(err)
# 放一个文件'application/csv'
try:
with open('./picture_files/long_lat.csv', 'rb') as file_data:
file_stat = ('./picture_files/long_lat.csv')
minioClient.put_object('pictures', 'long_lat.csv', file_data,
file_stat.st_size, content_type='application/csv')
print("Sussess")
except ResponseError as err:
print(err)
# 通过文件上传到对象中
def fput_object(self):
try:
print(minioClient.fput_object('pictures', '', './picture_files/'))
print("Sussess")
except ResponseError as err:
print(err)
try:
print(minioClient.fput_object('pictures', 'long_lat.csv',
'./picture_files/long_lat.csv',
content_type='application/csv'))
print("Sussess")
except ResponseError as err:
print(err)
# 获取对象的元数据
def stat_object(self):
try:
print(minioClient.stat_object('pictures', ''))
except ResponseError as err:
print(err)
# 删除对象
def remove_object(self):
try:
minioClient.remove_object('pictures', '')
print("Sussess")
except ResponseError as err:
print(err)
# 删除存储桶中的多个对象
def remove_objects(self):
try:
objects_to_delete = ['', 'long_lat.csv']
for del_err in minioClient.remove_objects('testfiles', objects_to_delete):
print("Deletion Error: {}".format(del_err))
print("Sussess")
except ResponseError as err:
print(err)
# 删除一个未完整上传的对象
def remove_incomplete_upload(self):
try:
minioClient.remove_incomplete_upload('testfiles', '')
print("Sussess")
except ResponseError as err:
print(err)
if __name__ == '__main__':
Object().remove_incomplete_upload()6、Python Presigned操作
from minio_learn.MinioConfig import minioClient
from import ResponseError
from datetime import timedelta
class Presigned:
# 生成一个用于HTTP GET操作的presigned URL
def presigned_get_object(self):
# 预先获得的对象名称的获取对象URL，将在2天后过期
try:
print(minioClient.presigned_get_object('testfiles', '', expires=timedelta(days=7)))
print("Sussess")
# 由于内部预定位确实会获得存储桶位置，因此仍然可能出现响应错误
except ResponseError as err:
print(err)
# 生成一个用于HTTP PUT操作的presigned URL
def presigned_put_object(self):
try:
print(minioClient.presigned_put_object('testfiles',
'',
expires=timedelta(days=7)))
print("Sussess")
except ResponseError as err:
print(err)
# 允许给POST操作的presigned URL设置策略条件。这些策略包括比如，
# 接收对象上传的存储桶名称，名称前缀，过期策略
def presigned_post_policy(self, PostPolicy):
pass
if __name__ == '__main__':
Presigned().presigned_put_object()Minio Cookbook 中文版链接：Minio SDKs - Python Client API文档 - 《Minio Cookbook 中文版》 - 书栈网 · BookStack