目录
本文介绍
MinIO 是一个开源的对象存储服务，它提供了简单而强大的 API，用于管理和操作存储桶中的对象。本文介绍了如何对 MinIO 的 API 进行封装，将其封装为易于使用的函数，并将封装后的函数上传到第三方库 Pyzjr 中。
上一节补充
使用官方的游乐场进行测试和开发
官方提供了一个MinIO服务器游乐场 https://play.min.io，可以随意使用此服务进行测试和开发。
好，现在我们先要用这个游乐场进行上传文件，熟悉操作。
from minio import Minio
from minio.error import S3Error
def main():
# Create a client with the MinIO server playground, its access key
# and secret key.
client = Minio(
"play.min.io",
access_key="Q3AM3UQ867SPQQA43P2F",
secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
)
# Make 'ikun' bucket if not exist.
found = client.bucket_exists("ikun")
if not found:
client.make_bucket("ikun")
else:
print("Bucket 'ikun' already exists")
client.fput_object(
"ikun", "ikun.mp4", "D:\Python_zjr\python_minio\ikun.mp4",
)
print(
"'D:\Python_zjr\python_minio\ikun.mp4' is successfully uploaded as "
"object 'ikun.mp4' to bucket 'ikun'."
)
if __name__ == "__main__":
try:
main()
except S3Error as exc:
print("error occurred.", exc)进入play.min.io，在搜索框中可以看见成功的创建了名为ikun的桶。点击进去，也发现了已经成功的上传了我们的视频文件。
熟悉MinIO的API
首先，我们了解了 MinIO 的基本概念和常用的 API 操作，包括创建存储桶、上传对象、下载对象、删除对象等。然后，通过使用 Python 的 MinIO 客户端库，我们将这些 API 操作封装为函数，使其更加易于使用和维护。封装的函数包括创建存储桶、上传对象、下载对象、删除对象等功能，并提供了一些灵活的参数选项。
创建客户端
from minio import Minio
from minio.error import S3Error
minioClient = Minio(
endpoint='play.minio.io:9000',
access_key='Q3AM3UQ867SPQQA43P2F',
secret_key='zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG'
)- endpoint：S3服务的主机名
- access_key：用户ID
- secret_key：密码
| 操作存储桶 | 操作对象 | Presigned操作 | 存储桶策略/通知 |
|---|---|---|---|
| make_bucket | get_object | presigned_get_object | get_bucket_policy |
| list_buckets | put_object | presigned_put_object | set_bucket_policy |
| bucket_exists | copy_object | presigned_post_policy | get_bucket_notification |
| remove_bucket | stat_object | set_bucket_notification | |
| list_objects | remove_object | remove_all_bucket_notification | |
| list_objects_v2 | remove_objects | listen_bucket_notification | |
| list_incomplete_uploads | remove_incomplete_upload | ||
| fput_object | |||
| fget_object | |||
| get_partial_object |
操作桶
1、检查桶是否存在，如果不存在就创建一个桶
def Foundbucket(client,bucket_name):
found = client.bucket_exists(bucket_name)
if not found:
client.make_bucket(bucket_name)
else:
print(f"Bucket {bucket_name} already exists")2、列出所有的存储桶名
def get_bucket_list(client):
try:
buckets = client.list_buckets()
for bucket in buckets:
print(bucket.name, bucket.creation_date) # 获取桶的名称和创建时间
except InvalidResponseError as err:
print(err)3、删除储存桶
def get_remove_bucket(client,bucket_name):
try:
client.remove_bucket(bucket_name)
print("删除存储桶成功")
except InvalidResponseError as err:
print(err)4、用于查看存储桶的对象
def get_bucket_files(client,bucket_name):
try:
objects = client.list_objects(bucket_name, prefix=None,
recursive=True)
for obj in objects:
print(obj.bucket_name, obj.object_name.encode('utf-8'), obj.last_modified,
obj.etag, obj.size, obj.content_type)
except InvalidResponseError as err:
print(err)操作对象
1、删除对象
def delete_object(client, bucket_name, objects):
try:
for obj in objects:
client.remove_object(bucket_name, obj)
print(f"Deleted {obj} under {bucket_name}")
except InvalidResponseError as err:
print(err)2、删除zip文件
def delete_folder(client, bucket_name, folder_path):
try:
temp_dir = tempfile.mkdtemp()
folder_name = os.path.basename(folder_path)
zip_file = os.path.join(temp_dir, f"{folder_name}.zip")
shutil.make_archive(zip_file[:-4], 'zip', folder_path)
client.remove_object(bucket_name, f"{folder_name}.zip")
print(f"Deleted {folder_name}.zip from bucket {bucket_name}")
shutil.rmtree(temp_dir)
except InvalidResponseError as err:
print(err)3、下载对象
def download_object(client, bucket_name, objects, filepath=None):
try:
if filepath is None:
current_dir = os.path.dirname(os.path.abspath(__file__)) # 获取当前文件的目录
else:
current_dir = filepath
for obj in objects:
file_path = os.path.join(current_dir, obj) # 拼接目录和文件名
client.fget_object(bucket_name, obj, file_path)
print(f"Downloaded object {obj} to {file_path}")
except InvalidResponseError as err:
print(err)4、下载zip文件，自动解压
def download_folder(client, bucket_name, local_path):
try:
temp_dir = tempfile.mkdtemp()
client.fget_object(bucket_name, f"{local_path}.zip", os.path.join(temp_dir, f"{local_path}.zip"))
print(f"Downloaded {local_path}.zip from bucket {bucket_name}")
shutil.unpack_archive(os.path.join(temp_dir, f"{local_path}.zip"), local_path)
print(f"Extracted {local_path}.zip to {local_path}")
shutil.rmtree(temp_dir)
except InvalidResponseError as err:
print(err)5、上传对象
def upload_object(client, bucket_name, file_path):
try:
object_name = os.path.basename(file_path) # 提取文件名作为对象名称
with open(file_path, "rb") as file_data:
file_size = os.path.getsize(file_path)
client.put_object(bucket_name, object_name, file_data, file_size)
print(f"Uploaded object {object_name} to bucket {bucket_name}")
except IOError as e:
print(f"Failed to open file: {file_path} - {e}")
except S3Error as err:
print(f"Error occurred: {err}")6、上传文件夹为zip文件
def upload_folder(client, bucket_name, folder_path):
try:
temp_dir = tempfile.mkdtemp()
folder_name = os.path.basename(folder_path)
zip_file = os.path.join(temp_dir, f"{folder_name}.zip")
shutil.make_archive(zip_file[:-4], 'zip', folder_path)
with open(zip_file, "rb") as file_data:
file_size = os.path.getsize(zip_file)
client.put_object(bucket_name, f"{folder_name}.zip", file_data, file_size)
print(f"Uploaded {folder_name}.zip to bucket {bucket_name}")
shutil.rmtree(temp_dir)
except IOError as e:
print(f"Failed to compress folder: {folder_path} - {e}")
except S3Error as err:
print(f"Error occurred: {err}")预签署
1、可供上传文件的URL，时限为2小时
def upload_url(self, client, bucket_name, object_name, expires_in=7200):
try:
# 生成预签名 URL
url = client.presigned_put_object(bucket_name, object_name, expires=timedelta(seconds=expires_in))
return url
except Exception as e:
print(f"Failed to generate presigned upload URL: {e}")
return None2、可供下载文件的URL，时限为2小时
def download_url(client, bucket_name, object_name, expires_in=7200):
try:
# 生成预签名 URL
url = client.presigned_get_object(bucket_name, object_name, expires=timedelta(seconds=expires_in))
return url
except Exception as e:
print(f"Failed to generate presigned download URL: {e}")
return None封装Minio的操作
第三方库 Pyzjr，它是一个开源的 Python 库。我将封装好的 MinIO API 函数上传到 Pyzjr 库中，可供我们团队轻松地使用这些函数，加速在 MinIO 上的开发工作。
1、先下载pyzjr第三方库
版本大于等于0.0.9是有对minio的封装的。
2、UML图
下面是pyzjr中对minio的封装，与上面的函数相同。
想要对MinIO进行更多的操作的话，可以查看github文档，里面提供了很多写好的示例，如果想要进行操作，可以先去看看里面的示例代码进行修改。我这里封装的是我认为可能会用到的，不完整但足够使用了。
参考文章
Python的API参考文档（英文与中文）：
Python Quickstart Guide — MinIO Object Storage for Linux
Minio SDKs - Python Client API文档 - 《Minio Cookbook 中文版》 - 书栈网 · BookStack
