标签：指南封包 Minio self bucket tag dict file name

【一】Docker部署Minio容器

【1】Minio容器介绍

Minio是一个开源的对象存储服务器，使用Apache License v2.0开源协议。
- 它提供了一个存储桶(bucket)的概念，类似于文件系统中的目录，用于存储对象文件。
- Minio与亚马逊S3云存储服务兼容，使得它可以轻松地与现有的S3应用程序集成。
Minio的优势在于其简洁和高性能。
- 它被设计成轻量级的，并且具有低延迟和高吞吐量的特点。
- 这使得Minio非常适合存储大容量的非结构化数据，例如图像、视频、日志文件、备份数据以及容器或虚拟机镜像等。
- 而且，Minio没有任何限制，一个对象文件可以是任意大小，从几KB到最大5TB不等。
Minio通过使用分布式架构来实现高可用性和冗余备份。
- 您可以配置多个Minio服务器，形成一个分布式集群来确保数据的安全性和可靠性。
- 此外，Minio还支持数据加密和访问控制，使您能够根据需要对数据进行保护和管理。
总之，Minio是一个功能强大且易于使用的对象存储服务器，适用于各种场景，包括数据存储、数据备份、容器存储等。
- 通过与亚马逊S3兼容的API，Minio能够轻松地与现有的应用程序集成，为您提供方便和高效的存储解决方案。

【2】寻找Minio镜像

Docker如果想安装软件 , 必须先到 Docker 镜像仓库下载镜像。

【3】下载Minio镜像

（1）下载最新版本镜像

docker pull minio/minio

下载最新版Minio镜像

其实此命令就等同于
docker pull minio/minio:latest

（2）下载指定版本镜像

docker pull minio/minio:RELEASE.2022-06-20T23-13-45Z.fips

下载指定版本的Minio镜像 (xxx指具体版本号)

（3）下载过程

镜像下载完成

Using default tag: latest
latest: Pulling from minio/minio
0c10cd59e10e: Pull complete 
ee2351f734c9: Pull complete 
6938b5623639: Pull complete 
52c7cc767226: Pull complete 
234b624c4f30: Pull complete 
67f2303e832e: Pull complete 
Digest: sha256:d6369c50d12a201b8fe72458c97fba79a50980184440b1e2fe93634dbf0ef682
Status: Downloaded newer image for minio/minio:latest
docker.io/minio/minio:latest

查看镜像
```
docker images
```

REPOSITORY           TAG       IMAGE ID       CREATED         SIZE
minio/minio          latest    b28473c6b8d0   2 days ago      276MB

【4】创建目录

一个用来存放配置，一个用来存储上传文件的目录

启动前需要先创建Minio外部挂载的配置文件（ /home/minio/config）和存储上传文件的目录（ /home/minio/data）

mkdir -p /data/minio/config
mkdir -p /data/minio/data

查看是否存在

# cd /data
# ls
minio
# cd minio
# ls
config  data

【5】创建Minio容器并运行

多行模式

docker run -p 9011:9011 -p 9010:9010 \
     --name minio \
     -d --restart=always \
     -e "MINIO_ACCESS_KEY=dream" \
     -e "MINIO_SECRET_KEY=dream521" \
     -v /data/minio/data:/data \
     -v /data/minio/config:/root/.minio \
     minio/minio server \
     /data --console-address ":9010" -address ":9011"

9090端口指的是minio的客户端端口

MINIO_ACCESS_KEY ：账号

MINIO_SECRET_KEY ：密码（账号长度必须大于等于5，密码长度必须大于等于8位）

配置文件路径要和上面的自定义路径一样

存储文件路径要和上面的自定义路径一样

单行模式

docker run -p 9010:9010 -p 9011:9011      --net=host      --name minio      -d --restart=always      -e "MINIO_ACCESS_KEY=dream"      -e "MINIO_SECRET_KEY=dream521"      -v /data/minio/data:/data      -v /data/minio/config:/root/.minio      minio/minio server      /data --console-address ":9010" -address ":9011"

【6】操作Minio

地址：http://IP:9090/login
登陆用户名：dream
登陆密码：dream521

（1）登录页面

（2）登陆成功

（3）创建用户

点击 Create User

输入用户名和密码

创建成功

（4）创建 Create Group

点击创建组

输入组的名字和组成员

创建完成

（5）创建accessKey和secretKey

点击 Create access key

会自动生成

点击Download for import

文件名：credentials.json
文件内容

{"url":"http://IP/api/v1/service-account-credentials","accessKey":"秘钥","secretKey":"秘钥","api":"s3v4","path":"auto"}

以上内容是自动生成的，我填文字的部分都是自动生成的

IP：自己的服务器IP

秘钥：上面自动生成的

秘钥：上面自动生成的

（6）创建桶(Bucket)

点击 Create Bucket

输入桶的名字

创建成功

（7）上传文件

查看已创建的桶

进入到已创建的桶内

点击 Upload
- 可以选择单文件/文件夹

上传文件成功

可以在这里查看桶的容量使用情况

【7】SDK操作

（1）相关链接

官方文档：https://docs.min.io/docs/

Java SDK：https://docs.min.io/docs/java-client-quickstart-guide.html

Python SDK:https://min.io/docs/minio/linux/developers/python/minio-py.html

（2）Java 操作上传文件

maven依赖

        #低版本的okhttp会报错提示
        <dependency>
            <groupId>com.squareup.okhttp3</groupId>
            <artifactId>okhttp</artifactId>
            <version>4.9.0</version>
        </dependency>
        
        <dependency>
            <groupId>io.minio</groupId>
            <artifactId>minio</artifactId>
            <version>8.4.2</version>
            <exclusions>
                <exclusion>
                    <artifactId>okhttp</artifactId>
                    <groupId>com.squareup.okhttp3</groupId>
                </exclusion>
            </exclusions>
        </dependency>

测试文件上传

import io.minio.BucketExistsArgs;
import io.minio.MakeBucketArgs;
import io.minio.MinioClient;
import io.minio.UploadObjectArgs;
import io.minio.errors.MinioException;
 
import java.io.IOException;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
 
public class FileUploader {
 
    public static void main(String[] args) throws IOException, NoSuchAlgorithmException, InvalidKeyException {
        try {
            // Create a minioClient with the MinIO server playground, its access key and secret key.
            MinioClient minioClient =
                    MinioClient.builder()
                            .endpoint("http://192.168.124.132:9000")
                            .credentials("XO1JDovW2FTmGaBb", "uG6wMfylUnOVH5WzwxqnldOWw2dMshNX")
                            .build();
 
            // Make 'asiatrip' bucket if not exist.
            boolean found = minioClient.bucketExists(BucketExistsArgs.builder().bucket("public").build());
            if (!found) {
                // Make a new bucket called 'asiatrip'.
                minioClient.makeBucket(MakeBucketArgs.builder().bucket("public").build());
            } else {
                System.out.println("Bucket 'public' already exists.");
            }
 
            // Upload '/home/user/Photos/asiaphotos.zip' as object name 'asiaphotos-2015.zip' to bucket
            // 'asiatrip'.
            minioClient.uploadObject(
                    UploadObjectArgs.builder()
                            .bucket("public")
                            .object("credentials.json")
                            .filename("C:/Users/lai.huanxiong/Downloads/credentials.json")
                            .build());
            System.out.println("'C:/Users/lai.huanxiong/Downloads/credentials.json' is successfully uploaded as " + "object 'credentials.json' to bucket 'public'.");
        } catch (MinioException e) {
            System.out.println("Error occurred: " + e);
            System.out.println("HTTP trace: " + e.httpTrace());
        }
    }
}

文件上传成功展示

（3）Python 操作上传文件

[1]最低要求

Python 3.7 或更高版本。

[2]安装第三方库

pip3 install minio

下载源码

git clone https://github.com/minio/minio-py
cd minio-py
python setup.py install

[3]快速入门示例

此示例程序连接到与 S3 兼容的对象存储服务器，在该服务器上创建一个存储桶，然后将文件上传到该存储桶。
您需要以下项目才能连接到与 S3 兼容的对象存储服务器：

参数	描述
端点	指向 S3 服务的网址。
访问密钥	S3 服务中账户的访问密钥（也称为用户 ID）。
密钥	S3 服务中账户的私有密钥（也称为密码）。

此示例使用 MinIO 服务器游乐场 https://play.min.io。
请随意使用此服务进行测试和开发。
file_uploader.py

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 'asiatrip' bucket if not exist.
    found = client.bucket_exists("asiatrip")
    if not found:
        client.make_bucket("asiatrip")
    else:
        print("Bucket 'asiatrip' already exists")

    # Upload '/home/user/Photos/asiaphotos.zip' as object name
    # 'asiaphotos-2015.zip' to bucket 'asiatrip'.
    client.fput_object(
        "asiatrip", "asiaphotos-2015.zip", "/home/user/Photos/asiaphotos.zip",
    )
    print(
        "'/home/user/Photos/asiaphotos.zip' is successfully uploaded as "
        "object 'asiaphotos-2015.zip' to bucket 'asiatrip'."
    )


if __name__ == "__main__":
    try:
        main()
    except S3Error as exc:
        print("error occurred.", exc)

运行文件上传程序

$ python file_uploader.py
'/home/user/Photos/asiaphotos.zip' is successfully uploaded as object 'asiaphotos-2015.zip' to bucket 'asiatrip'.

$ mc ls play/asiatrip/
[2016-06-02 18:10:29 PDT]  82KiB asiaphotos-2015.zip

【二】基于Python版本的SDK封装

特别感谢博主提供的思路支持

https://blog.csdn.net/weixin_40547993/article/details/110682587?ops_request_misc=&request_id=&biz_id=102&utm_term=PythonSDK

本次封装基于Minio官网提供的Python SDK 开发者文档进行二次封装

https://min.io/docs/minio/linux/developers/python/minio-py.html

【三】封装结构树

├── MinioPySdk
    ├── imgs				# README.md 用到的静态图片
    ├── __init__.py			# 初始化程序，可以将MinioPySdk直接导入，而不需要单独导入函数
    ├── py_sdk_minio.py    	# 项目主应用，开发时的代码保存
    ├── settings.py   		# 项目主应用配置文件(配置必填参数)
    ├── test.py  			# 项目主应用测试文件(含方法使用说明)
    ├── README.md			# 项目主应用介绍文件

__init__.py

# -*-coding: Utf-8 -*-
# @File : __init__ .py
# author: Chimengmeng
# blog_url : https://www.cnblogs.com/dream-ze/
# Time：2023/8/13
from .py_sdk_minio import *
from .settings import *

settings.py

# -*-coding: Utf-8 -*-
# @File : settings .py
# author: Chimengmeng
# blog_url : https://www.cnblogs.com/dream-ze/
# Time：2023/8/13

# endpoint指定的是你Minio的远程IP及端口
ENDPOINT = "43.138.48.158:9011"
# accesskey指定的是你的Minio服务器访问key
ACCESS_KEY = "7FGId3bewvgBq75Upf0Q"
# secret_key指定的是你登录时需要用的key，类似密码
SECRET_KEY = "5wvGpsF9f058sSBq9A3XJ06qQezLFlKnSu0vKNkh"
# secure指定是否以安全模式创建Minio连接,建议为False
SECURE = False
# 必须指定一个默认存储桶名称
BUCKET_NAME = 'testbucket'

py_sdk_minio.py

# -*-coding: Utf-8 -*-
# @File : py_sdk_minio .py
# author: Chimengmeng
# blog_url : https://www.cnblogs.com/dream-ze/
# Time：2023/8/13
import os
import pytz
from datetime import timedelta
from minio import Minio
from minio.commonconfig import CopySource
from minio.deleteobjects import DeleteObject
from minio.error import MinioException
from PIL import Image


# 桶相关操作
class BucketOperation(object):

    def __init__(self, endpoint, access_key, secret_key, bucket_name, secure=False):
        '''

        :param endpoint: IP:PORT(必传)
        :param access_key: 访问key,类似于用户名(必传)
        :param secret_key: 访问密钥,类似于密码(必传)
        :param bucket_name: 初识桶名(必传)
        :param secure: 以安全模式创建Minio连接,建议默认为 False (可选)
        :param back_dict: 反馈信息字典
                result : 执行成功/失败
                tag_bucket_name ：当前桶名
                msg ： 返回的单条消息
        '''
        self.endpoint = endpoint
        self.access_key = access_key
        self.secret_key = secret_key
        self.secure = secure
        self.bucket_name = bucket_name
        self.back_dict = {"result": True, "tag_bucket_name": "", "msg": "", "msg_list": []}
        self.minioClient = Minio(
            endpoint=endpoint,
            access_key=access_key,
            secret_key=secret_key,
            secure=secure
        )

    # 校验目标桶名
    def get_tag_bucket_name(self, tag_bucket_name=None):
        '''
        :param tag_bucket_name: 校验桶名
        :return: 返回桶名
        '''
        self.back_dict['tag_bucket_name'] = tag_bucket_name
        if tag_bucket_name:
            return tag_bucket_name
        else:
            return self.bucket_name

    # 创建桶 / 校验桶
    def create_bucket(self, tag_bucket_name=None):
        '''
        :param tag_bucket_name: 目标桶名,默认是初始化的桶名,可以自创建桶
        :return:
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            if self.minioClient.bucket_exists(bucket_name=tag_bucket_name):  # bucket_exists：检查桶是否存在
                self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 已经存在"
                return self.back_dict
            else:
                self.minioClient.make_bucket(tag_bucket_name)
                self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 创建成功"
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"初始化存储桶失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 列出所有的存储桶 list_buckets函数
    def get_bucket_list(self, tag_bucket_name=None):

        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            buckets = self.minioClient.list_buckets()
            for bucket in buckets:
                bucket_dict = {}
                bucket_dict['bucket_name'] = bucket.name
                bucket_create_time_utc = bucket.creation_date
                bucket_create_time_sh = bucket_create_time_utc.astimezone(pytz.timezone('Asia/Shanghai'))  # 获取中国北京时区的信息
                bucket_dict['bucket_create_time_utc'] = bucket_create_time_utc.strftime("%Y-%m-%d %H:%M:%S")
                bucket_dict['bucket_create_time_sh'] = bucket_create_time_sh.strftime("%Y-%m-%d %H:%M:%S")
                self.back_dict['msg_list'].append(bucket_dict)
            self.back_dict['msg'] = '存储桶列表获取成功'
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"获取桶列表失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 删除存储桶（单个存储桶）
    def get_remove_bucket(self, tag_bucket_name):
        '''

        :param tag_bucket_name: 当前桶名称
        :return: 返回信息字典
        '''
        try:
            self.minioClient.remove_bucket(tag_bucket_name)
            self.back_dict['tag_bucket_name'] = self.bucket_name
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 删除成功"
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 删除失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 列出目标存储桶中所有对象  或者使用 list_objects_v2也可
    def get_bucket_files(self, tag_bucket_name=None):
        '''

        :param tag_bucket_name: 当前桶名称
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            self.back_dict['msg_list'] = []
            object_dict = {}
            # prefix用于过滤的对象名称前缀
            # objects = self.minioClient.list_objects('my_bucket', prefix='photos/') #  列出指定前缀开头的对象
            # objects = self.minioClient.list_objects('my_bucket', prefix='photos/', recursive=False) # 仅列举指定前缀下的直接对象，不递归子目录
            objects = self.minioClient.list_objects(tag_bucket_name, prefix=None, recursive=True)
            for obj in objects:
                object_dict['obj_bucket_name'] = obj.bucket_name
                object_dict['obj_name'] = obj.object_name
                bucket_create_time_utc = obj.last_modified
                bucket_create_time_sh = bucket_create_time_utc.astimezone(pytz.timezone('Asia/Shanghai'))  # 获取中国北京时区的信息
                object_dict['last_modified_utc'] = bucket_create_time_utc.strftime("%Y-%m-%d %H:%M:%S")
                object_dict['last_modified_sh'] = bucket_create_time_sh.strftime("%Y-%m-%d %H:%M:%S")
                # etag : 对象的唯一标识符
                object_dict['etag'] = obj.etag
                object_dict['size'] = obj.size
                object_dict['content_type'] = obj.content_type
                self.back_dict['msg_list'].append(object_dict)
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 获取所有对象成功"
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 获取所有对象失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 列出存储桶中未完整上传的对象(废弃)
    def get_list_incomplete_uploads(self, tag_bucket_name=None):
        '''

        :param tag_bucket_name: 当前桶名称
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            uploads = self.minioClient.list_incomplete_uploads(tag_bucket_name, prefix=None, recursive=True)
            for obj in uploads:
                res = f'当前存储桶:>> {obj.bucket_name} 文件名:>> {obj.object_name} 上传id:>> {obj.upload_id} 文件大小:>> {obj.size}'
                print(res)
        except MinioException as err:
            res = f"温馨提示 : 出错啦!错误是 :>>{err}"
            print(res)

    # 获取存储桶的当前策略
    def bucket_policy(self, tag_bucket_name=None):
        '''

        :param tag_bucket_name: 当前桶名称
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            policy = self.minioClient.get_bucket_policy(tag_bucket_name)
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 获取存储策略成功"
            self.back_dict['policy'] = policy
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 获取存储策略失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 给指定的存储桶设置存储桶策略
    def get_set_bucket_policy(self, tag_bucket_name, policy):
        '''

        :param tag_bucket_name: 当前桶名称
        :param policy: 存储桶策略
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            '''
            # 示例 ： 
            policy = """
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {
                            "AWS": "arn:aws:iam::123456789012:user/Alice"
                        },
                        "Action": [
                            "s3:GetObject",
                            "s3:PutObject"
                        ],
                        "Resource": "arn:aws:s3:::test1/*"
                    }

                ]
            }
            """
            '''
            self.minioClient.set_bucket_policy(tag_bucket_name, policy)
            self.back_dict['msg'] = f"目标存储桶 存储策略 设置成功"
            # self.back_dict['policy'] = policy
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 设置存储策略失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 获取存储桶的通知配置（成功）
    def get_bucket_notification(self, tag_bucket_name=None):
        '''

        :param tag_bucket_name: 当前桶名称
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            notification = self.minioClient.get_bucket_notification(tag_bucket_name)
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 通知配置 获取成功"
            self.back_dict['notification'] = notification
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 通知配置 获取失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 设置存储桶的通知配置（失败）
    def re_set_bucket_notification(self, notification_config, tag_bucket_name=None):
        '''

        :param notification_config: 通知配置
        :param tag_bucket_name: 当前桶名称
        :return: 返回信息字典
        '''
        '''
        Example::
            config = NotificationConfig(
                queue_config_list=[
                    QueueConfig(
                        "QUEUE-ARN-OF-THIS-BUCKET",
                        ["s3:ObjectCreated:*"],
                        config_id="1",
                        prefix_filter_rule=PrefixFilterRule("abc"),
                    ),
                ],
            )
        :param notification_config: 通知配置
        :param tag_bucket_name: 目标桶
        :return:
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            # 设置存储桶的通知配置
            res = self.minioClient.set_bucket_notification(tag_bucket_name, config=notification_config)
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 通知配置 设置成功"
            self.back_dict['notification_config'] = notification_config
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['tag_bucket_name'] = tag_bucket_name
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 通知配置 设置失败"
            self.back_dict['error'] = err
            self.back_dict['notification_config'] = notification_config
        return self.back_dict

    # 移除存储桶上的所有通知（失败）
    def re_remove_all_bucket_notifications(self, tag_bucket_name=None):
        '''

        :param tag_bucket_name: 存储桶的名称
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            # 移除存储桶上的所有通知
            res = self.minioClient.delete_bucket_notification(tag_bucket_name)
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 通知配置 移除成功"
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 通知配置 移除失败"
            self.back_dict['error'] = err
        return self.back_dict

    # 监听存储桶的通知
    def listen_bucket_notification(self, prefix, suffix, events, tag_bucket_name=None):
        '''

        :param prefix: 对象前缀，用于匹配以指定前缀开头的对象。可选参数，默认为空字符串。
        :param suffix:对象后缀，用于匹配以指定后缀结尾的对象。可选参数，默认为空字符串。
        :param events: 需要监听的事件类型。可选参数，默认为`s3:ObjectCreated:*`、`s3:ObjectRemoved:*`、`s3:ObjectAccessed:*`三种事件类型。
        :param tag_bucket_name: 存储桶的名称
        :return: 返回信息字典
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            # 监听存储桶的通知
            for event in self.minioClient.listen_bucket_notification(tag_bucket_name, prefix=prefix, suffix=suffix,
                                                                     events=events):
                self.back_dict['result'] = True
                self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 监听成功"
                self.back_dict['event'] = event
                return self.back_dict
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_bucket_name} 监听失败"
            self.back_dict['error'] = err
            return self.back_dict


class FileOperation():
    def __init__(self, bucket_obj):
        '''

        :param bucket_obj: 存储桶对象 (必传)
        minioClient ： 父类的链接对象
        back_dict ： 返回信息字典
        get_tag_bucket_name : 校验当前桶对象
        '''
        self.minioClient = bucket_obj.minioClient
        self.bucket_name = bucket_obj.bucket_name
        self.back_dict = bucket_obj.back_dict
        self.get_tag_bucket_name = bucket_obj.get_tag_bucket_name

    # 从桶中下载一个对象（支持txt、csv文件）并保存到指定路径
    def load_object(self, save_file_path, tag_file_name, tag_bucket_name=None, start=0, end=0):
        '''

        :param save_file_path: 保存文件的路径
        :param tag_file_name: 桶名称
        :param tag_bucket_name: 文件名称
        :param start: 区间起始位置
        :param end:  区间结束位置
        :return: 下载成功或失败的提示信息
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            data = self.minioClient.get_object(tag_bucket_name, tag_file_name, start, end)  # 这里的data 是 WEBP 格式的数据
            type_save_file_path = os.path.join(save_file_path, tag_file_name)
            with open(type_save_file_path, 'wb') as file_data:
                for d in data.stream(32 * 1024):
                    file_data.write(d)
            type_file = tag_file_name.split('.')[-1].upper()
            if type_file in ['PNG', 'JPG', 'WEBP', 'JPEG', 'GIF', 'TIFF', 'BMP', 'SVG']:
                # 将WEBP图片转换为指定格式
                image = Image.open(type_save_file_path).convert('RGB')
                image.save(type_save_file_path, f"{type_file}")
            else:
                ...
            self.back_dict['msg'] = f"目标文件 {tag_file_name} 下载成功"
            return self.back_dict
        except Exception as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标存储桶 {tag_file_name} 下载失败"
            self.back_dict['error'] = err
            return self.back_dict


    # 下载并将文件保存到本地
    def fget_object(self, save_file_path, tag_file_name, tag_bucket_name=None):
        """

        :param save_file_path: 保存文件的路径
        :param tag_bucket_name: 桶名称
        :param tag_file_name:  文件名称
        :return:
        """
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            save_file_path = os.path.join(save_file_path, tag_file_name)
            res = self.minioClient.fget_object(tag_bucket_name, tag_file_name, save_file_path)
            self.back_dict['msg'] = f"目标文件下载成功"
            self.back_dict['addr'] = save_file_path
            return self.back_dict
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标文件 {tag_file_name} 下载失败"
            self.back_dict['error'] = err
            return self.back_dict

    # 拷贝对象存储服务上的源对象到一个新对象
    # 注：该API支持的最大文件大小是5GB
    # 可通过copy_conditions参数设置copy条件
    # 经测试copy复制28M的文件需要663ms; 1.8G的压缩包需要53s
    def get_copy_object(self, start_file_name, end_file_name=None, start_bucket_name=None, end_bucket_name=None):
        """
        拷贝对象存储服务上的源对象到一个新对象
        :param start_file_name: 源文件桶的文件名字
        :param start_bucket_name: 源文件桶对象
        :param end_file_name: 目标对象桶的文件名字
        :param end_bucket_name: 目标桶对象
        :param source_obj: source 对象 CopySource("my-sourcebucket", "my-sourceobject")
        :return:
        """
        tag_bucket_name = self.get_tag_bucket_name(end_bucket_name)
        if end_file_name:
            end_file_name = end_file_name
        else:
            end_file_name = start_file_name
        source_obj = CopySource(start_bucket_name, start_file_name)
        try:
            copy_result = self.minioClient.copy_object(tag_bucket_name, end_file_name, source_obj)
            self.back_dict['msg'] = f"源桶 :>> {start_bucket_name} 源文件 :>> {start_file_name} 目标桶 :>> {end_bucket_name} 目标文件 :>> {end_file_name} 拷贝成功"
            self.back_dict['copy_result'] = copy_result
            return self.back_dict
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标文件 {start_file_name} 拷贝失败"
            self.back_dict['error'] = err
            return self.back_dict

    # 添加一个新的对象到对象存储服务
    """
    单个对象的最大大小限制在5TB。put_object在对象大于5MiB时，自动使用multiple parts方式上传。
    这样，当上传失败时，客户端只需要上传未成功的部分即可（类似断点上传）。
    上传的对象使用MD5SUM签名进行完整性验证。
    """
    def upload_object(self, upload_file_path, tag_bucket_name=None, tag_file_name=None, content_type=None):
        '''

        :param upload_file_path: 本地上传文件的路径
        :param tag_bucket_name: 目标存储桶
        :param tag_file_name:  当前存储文件在存储桶的文件名
        :param content_type: 存储文件的类型(可根据需要选择) 指定上传的数据类型 "application/json"、"image/jpeg"、"text/plain"等。
        :return:
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        # 不选择类型
        if not content_type:
            try:
                with open(upload_file_path, 'rb') as file_data:
                    file_stat = os.stat(upload_file_path)
                    res = self.minioClient.put_object(tag_bucket_name, tag_file_name, file_data, file_stat.st_size)
                    download_url = self.minioClient.presigned_get_object(bucket_name=tag_bucket_name,
                                                                         object_name=tag_file_name,
                                                                         expires=timedelta(days=7),
                                                                         response_headers={
                                                                             'x-amz-server-side-encryption': 'AES256'})
                self.back_dict['msg'] = f"目标桶 {tag_bucket_name} 目标文件 {tag_file_name} 上传成功"
                self.back_dict['download_url'] = download_url
                return self.back_dict
            except MinioException as err:
                self.back_dict['result'] = False
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 上传失败"
                self.back_dict['error'] = err
                return self.back_dict
        # 指定其他类型
        else:
            try:
                with open(upload_file_path, 'rb') as file_data:
                    file_stat = os.stat(upload_file_path)
                    self.minioClient.put_object(tag_bucket_name, tag_file_name, file_data, file_stat.st_size,
                                                content_type=content_type)
                    download_url = self.minioClient.presigned_get_object(bucket_name=tag_bucket_name,
                                                                         object_name=tag_file_name,
                                                                         expires=timedelta(days=7),
                                                                         response_headers={
                                                                             'x-amz-server-side-encryption': 'AES256'})
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 文件类型 {content_type} 上传成功"
                self.back_dict['download_url'] = download_url
                return self.back_dict
            except MinioException as err:
                self.back_dict['result'] = False
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 文件类型 {content_type} 上传失败"
                self.back_dict['error'] = err
                return self.back_dict

    # 通过文件上传到对象中
    def fput_object(self, tag_file_name=None, tag_bucket_name=None, upload_file_path=None, content_type=None):
        '''

        :param upload_file_path: 本地上传文件的路径
        :param tag_bucket_name: 目标存储桶
        :param tag_file_name:  当前存储文件在存储桶的文件名
        :param content_type: 存储文件的类型(可根据需要选择) 指定上传的数据类型 "application/json"、"image/jpeg"、"text/plain"等。
        :return:
        '''
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        if not content_type:
            try:
                res = self.minioClient.fput_object(tag_bucket_name, tag_file_name, upload_file_path)
                # 返回一个下载链接 ： 有效期 7 天
                download_url = self.minioClient.presigned_get_object(bucket_name=tag_bucket_name,
                                                                     object_name=tag_file_name,
                                                                     expires=timedelta(days=7),
                                                                     response_headers={
                                                                         'x-amz-server-side-encryption': 'AES256'})
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 上传成功"
                self.back_dict['download_url'] = download_url
                return self.back_dict
            except MinioException as err:
                self.back_dict['result'] = False
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 上传失败"
                self.back_dict['error'] = err
                return self.back_dict
        else:
            try:
                res = self.minioClient.fput_object(tag_bucket_name, tag_file_name, upload_file_path,
                                                   content_type=content_type)
                download_url = self.minioClient.presigned_get_object(bucket_name=tag_bucket_name,
                                                                     object_name=tag_file_name,
                                                                     expires=timedelta(days=7),
                                                                     response_headers={
                                                                         'x-amz-server-side-encryption': 'AES256'})
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 文件类型 {content_type} 上传成功"
                self.back_dict['download_url'] = download_url
                return self.back_dict
            except MinioException as err:
                self.back_dict['result'] = False
                self.back_dict['msg'] = f"目标文件 {tag_file_name} 文件类型 {content_type} 上传失败"
                self.back_dict['error'] = err
                return self.back_dict

    # 获取对象的元数据
    def re_stat_object(self, tag_bucket_name=None, tag_file_name=None):
        '''

        :param tag_bucket_name: 目标存储桶
        :param tag_file_name:  当前存储文件在存储桶的文件名
        :return:
        '''

        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            stat = self.minioClient.stat_object(tag_bucket_name, tag_file_name)
            self.back_dict['msg'] = f"目标文件 {tag_file_name}  元数据获取成功"
            self.back_dict['stat'] = stat
            return self.back_dict
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标文件 {tag_file_name}  元数据获取失败"
            self.back_dict['error'] = err
            return self.back_dict

    # 删除对象
    def remove_object(self, tag_file_name=None, tag_bucket_name=None):
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)
        try:
            res = self.minioClient.remove_object(tag_bucket_name, tag_file_name)
            self.back_dict['msg'] = f"目标对象 {tag_file_name}  删除成功"
            return self.back_dict
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标对象 {tag_file_name}  删除失败"
            self.back_dict['error'] = err
            return self.back_dict

    # 删除存储桶中的多个对象
    def remove_objects(self, tag_bucket_name=None, tag_file_name_list=None):
        tag_bucket_name = self.get_tag_bucket_name(tag_bucket_name)

        try:
            # tag_file_name_list = ['1.txt', '2.txt', '3.txt']
            delete_object_list = []
            for tag_file_name in tag_file_name_list:
                delete_object_list.append(DeleteObject(tag_file_name))
            for del_err in self.minioClient.remove_objects(tag_bucket_name, delete_object_list):
                msg = f"目标桶 {tag_bucket_name}  目标文件 {del_err} 删除成功 "  # 源码里面会打印东西，但是我这里没法印出来，没搞懂
                print(msg)
            self.back_dict['msg'] = f"目标对象列表 {tag_file_name_list}  删除成功"
            return self.back_dict
        except MinioException as err:
            self.back_dict['result'] = False
            self.back_dict['msg'] = f"目标对象列表 {tag_file_name_list}  删除失败"
            self.back_dict['error'] = err
            return self.back_dict

    # 删除一个未完整上传的对象(失败)
    def remove_incomplete_upload(self, tag_bucket_name=None, tag_file_name=None):
        # 没有这个方法
        # 主动创建桶 tag_bucket_name 有值
        if tag_bucket_name:
            tag_bucket_name = tag_bucket_name
        else:
            tag_bucket_name = self.bucket_name
        try:
            self.minioClient.remove_incomplete_upload(tag_bucket_name, tag_file_name)
            msg = f"目标桶 {tag_bucket_name}  目标文件 {tag_file_name} 删除成功 "
            print(msg)
        except MinioException as err:
            msg = f"目标桶 {tag_bucket_name} 目标文件 {tag_file_name} 删除失败 失败原因 {err}"
            print(msg)


if __name__ == '__main__':
    ...

【四】存储桶相关说明

【1】初始化存储桶

（1）参数

参数名称	参数类型	注释	来源	权重
endpoint	字符串	endpoint指定的是你Minio的远程IP及端口	settings.ENDPOINT	必填参数
access_key	字符串	accesskey指定的是你的Minio服务器访问key	settings.ACCESS_KEY	必填参数
secret_key	字符串	secret_key指定的是你登录时需要用的key，类似密码	settings.SECRET_KEY	必填参数
bucket_name	字符串	必须指定一个默认存储桶名称	settings.BUCKET_NAME	必填参数
secure	字符串	secure指定是否以安全模式创建Minio连接,建议为False	settings.SECURE	必填参数

（2）实现

# 创建连接桶对象 __init__(self, endpoint, access_key, secret_key, bucket_name, secure=False)
bucket = BucketOperation(endpoint=settings.ENDPOINT, access_key=settings.ACCESS_KEY, secret_key=settings.SECRET_KEY,
                         bucket_name=settings.BUCKET_NAME, secure=settings.SECURE)

（3）结果

无返回结果

【2】检查桶(默认不创建)

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	非必传参数

（2）实现

# 检查桶，默认不创建 -- create_bucket(self, tag_bucket_name=None)
bucket.create_bucket()

（3）结果

{'result': True, 'tag_bucket_name': None, 'msg': '目标存储桶 testbucket 已经存在', 'msg_list': []}

【3】创建桶(主动创建新桶)

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	必传参数

（2）实现

# 创建桶，主动创建 -- create_bucket(self, tag_bucket_name=None)
tag_bucket_name = 'test5'
bucket.create_bucket(tag_bucket_name=tag_bucket_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test5', 'msg': '目标存储桶 test5 创建成功', 'msg_list': []}

【4】获取所有桶列表

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	可选参数

（2）实现

# 获取所有桶列表 -- get_bucket_list(self, tag_bucket_name=None)
bucket.get_bucket_list()

（3）结果

{'result': True, 'tag_bucket_name': None, 'msg': '存储桶列表获取成功', 'msg_list': [
    {'bucket_name': 'test1', 'bucket_create_time_utc': '2023-08-12 08:55:10','bucket_create_time_sh': '2023-08-12 16:55:10'},
    {'bucket_name': 'test2', 'bucket_create_time_utc': '2023-08-12 13:19:30','bucket_create_time_sh': '2023-08-12 21:19:30'},
    {'bucket_name': 'test4', 'bucket_create_time_utc': '2023-08-13 03:19:34','bucket_create_time_sh': '2023-08-13 11:19:34'},
    {'bucket_name': 'test5', 'bucket_create_time_utc': '2023-08-13 03:19:57','bucket_create_time_sh': '2023-08-13 11:19:57'},
    {'bucket_name': 'testbucket', 'bucket_create_time_utc': '2023-08-12 05:18:30','bucket_create_time_sh': '2023-08-12 13:18:30'}
]}

【5】指定存储桶访问

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	可选参数

（2）实现

# 指定存储桶访问 -- get_bucket_list(self, tag_bucket_name=None)
tag_bucket_name = 'test1'
bucket.get_bucket_list(tag_bucket_name=tag_bucket_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test1', 'msg': '存储桶列表获取成功', 'msg_list': [
    {'bucket_name': 'test1', 'bucket_create_time_utc': '2023-08-12 08:55:10','bucket_create_time_sh': '2023-08-12 16:55:10'},
    {'bucket_name': 'test2', 'bucket_create_time_utc': '2023-08-12 13:19:30','bucket_create_time_sh': '2023-08-12 21:19:30'},
    {'bucket_name': 'test4', 'bucket_create_time_utc': '2023-08-13 03:19:34','bucket_create_time_sh': '2023-08-13 11:19:34'},
    {'bucket_name': 'test5', 'bucket_create_time_utc': '2023-08-13 03:19:57','bucket_create_time_sh': '2023-08-13 11:19:57'},
    {'bucket_name': 'testbucket', 'bucket_create_time_utc': '2023-08-12 05:18:30','bucket_create_time_sh': '2023-08-12 13:18:30'}
]}

【6】删除指定桶

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	必传参数

（2）实现

# 删除指定桶 -- get_remove_bucket(self, tag_bucket_name=None)
tag_bucket_name = 'test5'
bucket.get_remove_bucket(tag_bucket_name=tag_bucket_name)

（3）结果

无返回结果，但是数据库发生变化，指定桶被删除

【7】列出目标存储桶中所有对象

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)

（2）实现

默认桶

# 列出目标存储桶中所有对象 -- get_bucket_files(self, tag_bucket_name)
bucket.get_bucket_files()

携带指定桶

tag_bucket_name = 'test2'
bucket.get_bucket_files(tag_bucket_name=tag_bucket_name)

（3）结果

默认桶

{'result': True, 'tag_bucket_name': None, 'msg': '目标存储桶 testbucket 获取所有对象成功', 'msg_list': 
    [{'obj_bucket_name': 'testbucket', 'obj_name': '驿站.png', 'last_modified_utc': '2023-08-13 03:52:26',
      'last_modified_sh': '2023-08-13 11:52:26', 'etag': '09ecc593ce0d0228a9657a85863ed0be', 'size': 6522,
      'content_type': None},
     {'obj_bucket_name': 'testbucket', 'obj_name': '驿站.png', 'last_modified_utc': '2023-08-13 03:52:26',
      'last_modified_sh': '2023-08-13 11:52:26', 'etag': '09ecc593ce0d0228a9657a85863ed0be', 'size': 6522,
      'content_type': None},
     {'obj_bucket_name': 'testbucket', 'obj_name': '驿站.png', 'last_modified_utc': '2023-08-13 03:52:26',
      'last_modified_sh': '2023-08-13 11:52:26', 'etag': '09ecc593ce0d0228a9657a85863ed0be', 'size': 6522,
      'content_type': None},
     {'obj_bucket_name': 'testbucket', 'obj_name': '驿站.png', 'last_modified_utc': '2023-08-13 03:52:26',
      'last_modified_sh': '2023-08-13 11:52:26', 'etag': '09ecc593ce0d0228a9657a85863ed0be', 'size': 6522,
      'content_type': None}
]}

指定桶返回结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': '目标存储桶 test2 获取所有对象成功', 'msg_list': 
    [{'obj_bucket_name': 'test2', 'obj_name': '2222.png', 'last_modified_utc': '2023-08-12 14:07:20',
      'last_modified_sh': '2023-08-12 22:07:20', 'etag': '7603d74c4e836fde0cdf895c172ad7fb', 'size': 20912,
      'content_type': None},
     {'obj_bucket_name': 'test2', 'obj_name': '2222.png', 'last_modified_utc': '2023-08-12 14:07:20',
      'last_modified_sh': '2023-08-12 22:07:20', 'etag': '7603d74c4e836fde0cdf895c172ad7fb', 'size': 20912,
      'content_type': None}
]}

【8】列出存储桶中未完整上传的对象(失败)

测试失败(废除)

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)

（2）实现

# 列出存储桶中未完整上传的对象(废除) -- get_list_incomplete_uploads(self, tag_bucket_name=None)
tag_bucket_name = 'test5'
bucket.get_list_incomplete_uploads(tag_bucket_name=tag_bucket_name)

（3）结果

【9】给指定的存储桶设置存储桶策略

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)
policy	字符串	存储桶策略	必传参数

（2）实现

policy示例

policy = """
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {
                            "AWS": "arn:aws:iam::123456789012:user/Alice"
                        },
                        "Action": [
                            "s3:GetObject",
                            "s3:PutObject"
                        ],
                        "Resource": "arn:aws:s3:::test1/*"
                    }

                ]
            }
            """

# 给指定的存储桶设置存储桶策略 -- get_set_bucket_policy(self, tag_bucket_name, policy)
policy = """
            {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Principal": {
                            "AWS": "arn:aws:iam::123456789012:user/Alice"
                        },
                        "Action": [
                            "s3:GetObject",
                            "s3:PutObject"
                        ],
                        "Resource": "arn:aws:s3:::test1/*"
                    }

                ]
            }
            """
tag_bucket_name = 'test2'
bucket.get_set_bucket_policy(tag_bucket_name=tag_bucket_name, policy=policy)

（3）结果

成功

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标存储桶 存储策略 设置成功', 'msg_list': []}

失败

{'result': False, 'tag_bucket_name': 'test2', 'msg': '目标存储桶 test2 设置存储策略失败', 'msg_list': [], 'error': S3Error('S3 operation failed; code: MalformedPolicy, message: bucket name does not match, resource: /test2, request_id: 177AD5BFADE2D9ED, host_id: 901192593c8d4aebefcd5e1ffb207b1846522d03b1e9cb2132c2770ef491581c, bucket_name: test2')}

【10】获取存储桶的当前策略

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)

（2）实现

# 获取存储桶的当前策略 -- bucket_policy(self, tag_bucket_name=None)
tag_bucket_name = 'test1'
bucket.bucket_policy(tag_bucket_name=tag_bucket_name)

（3）结果

成功

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标存储桶 test1 获取存储策略成功', 'msg_list': [], 'policy': '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"AWS":["arn:aws:iam::123456789012:user/Alice"]},"Action":["s3:PutObject","s3:GetObject"],"Resource":["arn:aws:s3:::test1/*"]}]}'}

失败

{'result': False, 'tag_bucket_name': 'test2', 'msg': '目标存储桶 test2 获取存储策略失败', 'msg_list': [], 'error': S3Error('S3 operation failed; code: NoSuchBucketPolicy, message: The bucket policy does not exist, resource: /test2, request_id: 177AD5F177D00E9A, host_id: 901192593c8d4aebefcd5e1ffb207b1846522d03b1e9cb2132c2770ef491581c, bucket_name: test2')}

【11】获取存储桶的通知配置

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)

（2）实现

# 获取存储桶的通知配置 -- get_bucket_notification(self, tag_bucket_name=None)
tag_bucket_name = 'test1'
bucket.get_bucket_notification(tag_bucket_name=tag_bucket_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标存储桶 test1 通知配置 获取成功', 'msg_list': [], 'notification': <minio.notificationconfig.NotificationConfig object at 0x00000253A47F3A60>}

【12】设置存储桶的通知配置(失败)

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)
notification_config	NotificationConfig类型	通知配置	必传参数

（2）实现

# 设置存储桶的通知配置 -- re_set_bucket_notification(self, notification_config, tag_bucket_name=None)
config = NotificationConfig(
    queue_config_list=[
        QueueConfig(
            "QUEUE-ARN-OF-THIS-BUCKET",
            ["s3:ObjectCreated:*"],
            config_id="1",
            prefix_filter_rule=PrefixFilterRule("abc"),
        ),
    ],
)
tag_bucket_name = 'test1'
bucket.re_set_bucket_notification(tag_bucket_name=tag_bucket_name,notification_config=config)

（3）结果

{'result': False, 'tag_bucket_name': 'test1', 'msg': '目标存储桶 test1 通知配置 设置失败', 'msg_list': [], 'error': S3Error('S3 operation failed; code: InvalidArgument, message: A specified destination ARN does not exist or is not well-formed. Verify the destination ARN., resource: /test1, request_id: 177AD65BCF1969A7, host_id: 901192593c8d4aebefcd5e1ffb207b1846522d03b1e9cb2132c2770ef491581c, bucket_name: test1'), 'notification_config': <minio.notificationconfig.NotificationConfig object at 0x000001DB32814460>}

【13】移除存储桶上的所有通知配置

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)

（2）实现

# 移除存储桶上的所有通知 -- re_remove_all_bucket_notifications(self, tag_bucket_name=None)
tag_bucket_name = 'test1'
bucket.re_remove_all_bucket_notifications(tag_bucket_name=tag_bucket_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标存储桶 test1 通知配置 移除成功', 'msg_list': []}

【14】监听存储桶的通知

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)
prefix	字符串	对象前缀，用于匹配以指定前缀开头的对象。可选参数，默认为空字符串。	可选参数
suffix	字符串	对象后缀，用于匹配以指定后缀结尾的对象。可选参数，默认为空字符串。	可选参数
events	字符串/元祖	需要监听的事件类型。可选参数，默认为`s3:ObjectCreated:`、`s3:ObjectRemoved:`、`s3:ObjectAccessed:*`三种事件类型。	可选参数

（2）实现

# 监听存储桶的通知 -- listen_bucket_notification(self, prefix, suffix, events, tag_bucket_name=None)

prefix = 'documents/'
suffix = '.pdf'
events = ('s3:ObjectCreated:*',
          's3:ObjectRemoved:*',
          's3:ObjectAccessed:*')
tag_bucket_name = 'test1'

bucket.listen_bucket_notification(tag_bucket_name=tag_bucket_name, prefix=prefix, suffix=suffix, events=events)

（3）结果

夯住了，我猜是处于监听状态了

【五】文件操作

【1】初始化文件操作对象

（1）参数

参数名称	参数类型	注释	权重
bucket	bucket对象	已经初始化的bucket对象	必传参数

（2）实现

# 创建文件操作对象 -- __init__(self, bucket_obj)

file = FileOperation(bucket)

（3）结果

无返回结果

【2】从桶中下载一个对象(下载文件)

get_object 也是一种获取对象的方法，用于从内存或缓存中获取对象数据。

典型的应用场景是通过缓存机制将对象数据缓存在内存中，提高读取性能。

对于频繁需要读取的对象数据，可以使用 get_object 方法从缓存中快速获取，减少对底层数据源的访问。

get_object 方法一般直接从内存或缓存中获取对象，不涉及底层数据源的访问。

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)
save_file_path	字符串	保存文件的路径	必传参数
tag_file_name	字符串	桶中存在的文件名称	必传参数
start	数字	区间起始位置	默认是0(下载完成文件)
end	字符串	区间结束位置	默认是0(下载完成文件)

（2）实现

# 从桶中下载一个对象（支持txt、csv文件）并保存到指定路径 -- load_object(self, save_file_path, tag_file_name, tag_bucket_name=None, start=0, end=0)

save_file_path = r'E:\Old Boy\luffy\luffyCity\luffyCity\libs\MinioPySdk'
tag_file_name = '4.png'
# tag_file_name = '02.json'
tag_bucket_name = 'test1'
file.load_object(save_file_path=save_file_path,tag_file_name=tag_file_name,tag_bucket_name=tag_bucket_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标文件 02.json 下载成功', 'msg_list': []}

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标文件 4.png 下载成功', 'msg_list': []}

【3】下载并将文件保存到本地(下载文件)

fget_object 是一种获取对象的方法，用于从某个数据源或存储系统中获取对象数据。

典型的应用场景是从数据库、文件系统或网络资源中获取对象的数据。

通常需要提供参数来指定数据源、路径或关键字等，以便获取特定的对象数据。

fget_object 方法一般会封装访问数据源的底层逻辑，如打开文件、查询数据库等。

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	存储桶名称	选传参数(不传则是默认存储桶)
save_file_path	字符串	保存文件的路径	必传参数
tag_file_name	字符串	桶中存在的文件名称	必传参数

（2）实现

# 下载并将文件保存到本地 -- fget_object(self, save_file_path, tag_file_name, tag_bucket_name=None)

save_file_path = r'E:\Old Boy\luffy\luffyCity\luffyCity\libs\MinioPySdk'
tag_file_name = '4.png'
tag_bucket_name = 'test1'
file.fget_object(save_file_path=save_file_path, tag_file_name=tag_file_name, tag_bucket_name=tag_bucket_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test1', 'msg': '目标文件下载成功', 'msg_list': [], 'addr': 'E:\\Old Boy\\luffy\\luffyCity\\luffyCity\\libs\\MinioPySdk\\4.png'}

【4】拷贝对象存储服务上的源对象到一个新对象(复制文件)

（1）参数

参数名称	参数类型	注释	权重
start_file_name	字符串	源文件桶的文件名字	必填(不填就是默认桶中存在的文件)
start_bucket_name	字符串	源文件桶对象	必填(不填就是默认桶)
end_file_name	字符串	目标对象桶的文件名字	必填
end_bucket_name	字符串	目标桶对象	必填

（2）实现

# 拷贝对象存储服务上的源对象到一个新对象 -- get_copy_object(self, start_file_name, end_file_name=None, start_bucket_name=None, end_bucket_name=None)

start_bucket_name = 'test4'
start_file_name = '我的.png'

end_bucket_name = 'test2'
end_file_name = '我的test2.png'
file.get_copy_object(start_bucket_name=start_bucket_name, start_file_name=start_file_name,
                           end_bucket_name=end_bucket_name, end_file_name=end_file_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': '源桶 :>> test4 源文件 :>> 我的.png 目标桶 :>> test2 目标文件 :>> 我的test2.png 拷贝成功', 'msg_list': [], 'copy_result': <minio.helpers.ObjectWriteResult object at 0x000001A27A2124F0>}

【5】添加一个新的对象到对象存储服务(上传文件)

put_object是另一种常见的上传文件方法，通常用于单个小文件或者网络非常稳定的情况。

使用put_object方法时，会将整个文件一次性上传到云端存储空间。

如果在上传过程中网络连接中断，put_object方法无法自动恢复上传，需要重新启动上传过程。

对于小文件或网络稳定的情况，使用put_object方法可以简单方便地将文件上传到云端存储空间。

（1）参数

参数名称	参数类型	注释	权重
upload_file_path	字符串	本地上传文件的路径	必填(不填就是默认桶中存在的文件)
tag_bucket_name	字符串	目标存储桶	必填(不填就是默认桶)
tag_file_name	字符串	当前存储文件在存储桶的文件名	必填
content_type	字符串	存储文件的类型(可根据需要选择) 指定上传的数据类型 "application/json"、"image/jpeg"、"text/plain"等。	可选

（2）实现

# 添加一个新的对象到对象存储服务 -- upload_object(self, upload_file_path, tag_bucket_name=None, tag_file_name=None, content_type=None)

upload_file_path = r'E:\Old Boy\luffy\luffyCity\luffyCity\libs\MinioPySdk\4.png'
tag_bucket_name = 'test2'
tag_file_name = '4.png'
content_type = ''
file.upload_object(upload_file_path=upload_file_path, tag_bucket_name=tag_bucket_name,
                         tag_file_name=tag_file_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': '目标桶 test2 目标文件 4.png 上传成功', 'msg_list': [], 'download_url': 'http://43.138.48.158:9011/test2/4.png?x-amz-server-side-encryption=AES256&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=7FGId3bewvgBq75Upf0Q%2F20230813%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20230813T065837Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=86aaad300db93f2a90abd05c25245f9031c15e787cdacf9d47d93bf953a46e9f'}

【6】通过文件上传到对象中(上传文件)

fput_object是芝浩科技开发的一种对象存储SDK提供的功能，其主要特点是支持断点续传。

在使用fput_object方法时，可以将待上传的文件拆分成多个分片进行传输，以实现在网络不稳定的情况下，更可靠地上传大文件。

如果上传过程中网络连接中断，fput_object方法可以自动寻找中断点，并从中断处继续上传，而无需重新上传整个文件。

使用fput_object方法需要在SDK中配置分片大小、分片重试次数等参数。

对于较大的文件，特别是在网络环境不稳定的情况下，建议使用fput_object方法进行上传，以实现断点续传，保证文件上传的可靠性和稳定性。

（1）参数

参数名称	参数类型	注释	权重
upload_file_path	字符串	本地上传文件的路径	必填(不填就是默认桶中存在的文件)
tag_bucket_name	字符串	目标存储桶	必填(不填就是默认桶)
tag_file_name	字符串	当前存储文件在存储桶的文件名	必填
content_type	字符串	存储文件的类型(可根据需要选择) 指定上传的数据类型 "application/json"、"image/jpeg"、"text/plain"等。	可选

（2）实现

# 通过文件上传到对象中 -- fput_object(self, tag_file_name=None, tag_bucket_name=None, upload_file_path=None, content_type=None)

upload_file_path = r'E:\Old Boy\luffy\luffyCity\luffyCity\libs\MinioPySdk\4.png'
tag_bucket_name = 'test2'
tag_file_name = '4.png'
content_type = ''
file.upload_object(upload_file_path=upload_file_path, tag_bucket_name=tag_bucket_name,
                         tag_file_name=tag_file_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': '目标桶 test2 目标文件 4.png 上传成功', 'msg_list': [], 'download_url': 'http://43.138.48.158:9011/test2/4.png?x-amz-server-side-encryption=AES256&X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=7FGId3bewvgBq75Upf0Q%2F20230813%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20230813T065837Z&X-Amz-Expires=604800&X-Amz-SignedHeaders=host&X-Amz-Signature=86aaad300db93f2a90abd05c25245f9031c15e787cdacf9d47d93bf953a46e9f'}

【7】获取对象的元数据

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	目标存储桶	必填(不填就是默认桶)
tag_file_name	字符串	当前存储文件在存储桶的文件名	必填

（2）实现

# 获取对象的元数据 -- re_stat_object(self, tag_bucket_name=None, tag_file_name=None)

tag_bucket_name = 'test2'
tag_file_name = '4.png'
file.re_stat_object( tag_bucket_name=tag_bucket_name,tag_file_name=tag_file_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': '目标文件 4.png  元数据获取成功', 'msg_list': [], 'stat': <minio.datatypes.Object object at 0x000001B8572023D0>}

【8】删除存储桶中的对象(单个)

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	目标存储桶	必填(不填就是默认桶)
tag_file_name	字符串	当前存储文件在存储桶的文件名	必填

（2）实现

# 删除对象 -- remove_object(self, tag_file_name=None, tag_bucket_name=None)

tag_bucket_name = 'test2'
tag_file_name = '4.png'
file.remove_object(tag_bucket_name=tag_bucket_name, tag_file_name=tag_file_name)

（3）结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': '目标对象 4.png  删除成功', 'msg_list': []}

【9】删除存储桶中的多个对象(多个)

（1）参数

参数名称	参数类型	注释	权重
tag_bucket_name	字符串	目标存储桶	必填(不填就是默认桶中存在的文件)
tag_file_name_list	列表	当前存储文件在存储桶的文件名列表	必填

（2）实现

# 删除存储桶中的多个对象 -- remove_objects(self, tag_bucket_name=None, tag_file_name_list=None)

tag_bucket_name = 'test2'
tag_file_name_list = ['1.png', '2.png', '3.png']
file.remove_objects(tag_bucket_name=tag_bucket_name, tag_file_name_list=tag_file_name_list)

（3）结果

{'result': True, 'tag_bucket_name': 'test2', 'msg': "目标对象列表 ['1.png', '2.png', '3.png']  删除成功", 'msg_list': []}

【4】删除一个未完整上传的对象(失败)

（1）参数

参数名称	参数类型	注释	权重

（2）实现

（3）结果

标签：指南,封包,Minio,self,bucket,tag,dict,file,name
From： https://www.cnblogs.com/dream-ze/p/17642805.html

【补充】Minio存储桶封包指南

【一】Docker部署Minio容器

【1】Minio容器介绍

【2】寻找Minio镜像

【3】下载Minio镜像

（1）下载最新版本镜像

（2）下载指定版本镜像

（3）下载过程

【4】创建目录

【5】创建Minio容器并运行

【6】操作Minio

（1）登录页面

（2）登陆成功

（3）创建用户

（4）创建 Create Group

（5）创建accessKey和secretKey

（6）创建桶(Bucket)

（7）上传文件

【7】SDK操作

（1）相关链接

（2）Java 操作上传文件

（3）Python 操作上传文件

[1]最低要求

[2]安装第三方库

[3]快速入门示例

【二】基于Python版本的SDK封装

【三】封装结构树

【四】存储桶相关说明

【1】初始化存储桶

（1）参数

（2）实现

（3）结果

【2】检查桶(默认不创建)

（1）参数

（2）实现

（3）结果

【3】创建桶(主动创建新桶)

（1）参数

（2）实现

（3）结果

【4】获取所有桶列表

（1）参数

（2）实现

（3）结果

【5】指定存储桶访问

（1）参数

（2）实现

（3）结果

【6】删除指定桶

（1）参数

（2）实现

（3）结果

【7】列出目标存储桶中所有对象

（1）参数

（2）实现

（3）结果

【8】列出存储桶中未完整上传的对象(失败)

（1）参数

（2）实现

（3）结果

【9】给指定的存储桶设置存储桶策略

（1）参数

（2）实现

（3）结果

【10】获取存储桶的当前策略

（1）参数

（2）实现

（3）结果

【11】获取存储桶的通知配置

（1）参数

（2）实现

（3）结果

【12】设置存储桶的通知配置(失败)

（1）参数

（2）实现

（3）结果

【13】移除存储桶上的所有通知配置

（1）参数

（2）实现

（3）结果