minio

1. 使用下列文件开启服务

minio.exe

2. 修改密码及开启服务

第一步，找到minio.exe所在目录，进入cmd窗口。

第二步，输入set MINIO_ACCESS_KEY=admin命令，修改ACCESS-KEY。

第三步，输入set MINIO_SECRET_KEY=12345678命令，修改 SECRET-KEY。

第四步，输入minio.exe server D:\minio，启动minio。

3.说明

4.操作类

minio_oss.py

import io
import re
import os
import logging
import mimetypes
import urllib.parse

from minio import Minio
from io import IOBase

_logger = logging.getLogger(__name__)


class BaseMinioError(Exception):
    code = "BaseMinioError"
    message = ""

    def __init__(self, func, *args, **kwargs):
        self.func = func.func.__name__  # 偏函数.func.__main__ 是实际调用的对象名称
        self.args = args
        self.kwargs = kwargs

    def __str__(self):
        return f"{self.code}:{self.message},func:{self.func},args:{self.args},kwargs:{self.kwargs}"


class ObjectTypeError(BaseMinioError):
    code = "ObjectTypeError"
    message = "The data must be 'str'/'bytes'/'IO object' "

    def __init__(self):
        pass

    def __str__(self):
        return f"{self.code}:{self.message}"


class NoSuchKey(BaseMinioError):
    code = "NoSuchKey"
    message = "Object does not exist"


class DataDecorator(object):
    def __init__(self, data):
        try:
            self.data = data.encode("utf8")
        except AttributeError:
            self.data = data

    def read(self, size: int = 0):
        if not size:
            return self.data
        return self.data[0:size]


_EXTRA_TYPES_MAP = {
    ".js": "application/javascript",
    ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    ".xltx": "application/vnd.openxmlformats-officedocument.spreadsheetml.template",
    ".potx": "application/vnd.openxmlformats-officedocument.presentationml.template",
    ".ppsx": "application/vnd.openxmlformats-officedocument.presentationml.slideshow",
    ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
    ".sldx": "application/vnd.openxmlformats-officedocument.presentationml.slide",
    ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    ".dotx": "application/vnd.openxmlformats-officedocument.wordprocessingml.template",
    ".xlam": "application/vnd.ms-excel.addin.macroEnabled.12",
    ".xlsb": "application/vnd.ms-excel.sheet.binary.macroEnabled.12",
    ".apk": "application/vnd.android.package-archive",
    # ".pdf": "application/pdf",
}


def content_type_by_name(name):
    """根据文件名，返回Content-Type。"""
    ext = os.path.splitext(name)[1].lower()
    if ext in _EXTRA_TYPES_MAP:
        return _EXTRA_TYPES_MAP[ext]

    return mimetypes.guess_type(name)[0]


class MinioOss:
    def __init__(self, host, access_key, secret_key, bucket, model_dirs):
        self.host = host
        self.access_key = access_key
        self.secret_key = secret_key
        self.secure = False
        self.bucket = bucket
        self.model_dirs = model_dirs
        self.client = Minio(
            self.host,
            secure=self.secure,
            access_key=self.access_key,
            secret_key=self.secret_key,
        )

    def __new__(cls, *args, **kw):
        """
        单例
        """
        if not hasattr(cls, 'instance'):
            cls.instance = super(MinioOss, cls).__new__(cls)
        return cls.instance

    def parse_path(self, path):
        """
        @param path:文件路径
        @return:
        """
        if path.startswith("/"):
            return f"{self.model_dirs}{path}"
        else:
            return f"{self.model_dirs}/{path}"

    def object_exists(self, bucket_name, path: str) -> bool:
        """
        判断文件是否存在
        @param path: 文件路径
        @return:
        """
        path = self.parse_path(path)

        try:
            self.client.stat_object(bucket_name, path)
        except NoSuchKey:
            return False

        return True

    def upload(self, object_name, raw_data, length):
        """
        上传IO流到Minio
        @param object_name: 对象在桶中的名字
        @param raw_data: An object having callable read() returning bytes object
        @param expire: 过期时间 单位：s
        @param length: 文件大小
        @return:
        """
        object_name = self.parse_path(object_name)
        # 当上传数据不是 str bytes 和IO对象时报错
        if not isinstance(raw_data, (IOBase, str, bytes)):
            raise ObjectTypeError()

        if not isinstance(raw_data, IOBase):
            raw_data = DataDecorator(raw_data)

        found = self.client.bucket_exists(self.bucket)
        if not found:
            self.client.make_bucket(self.bucket)
            _logger.info(f'create {self.bucket} sucess！')

        return self.client.put_object(
            bucket_name=self.bucket, object_name=object_name, data=raw_data, length=length,
            content_type=content_type_by_name(object_name)
        )

    def upload_file(self, object_name, file_name):
        """
        上传文件到Minio
        @param object_name: 桶里面的对象名
        @param file_name: 上传的文件名
        @return:
        """
        object_name = self.parse_path(object_name)

        found = self.client.bucket_exists(self.bucket)
        if not found:
            self.client.make_bucket(self.bucket)
            _logger.info(f'create {self.bucket} sucess！')

        self.client.fput_object(
            self.bucket, object_name, file_name, content_type=content_type_by_name(object_name))

    def download_url(self, object_name):
        """
        获取minio文件下载地址
        @param object_name: 桶里面的对象名
        @return:
        """
        if re.findall('[\u4e00-\u9fa5]', object_name):
            # 当文件名为中文时，minio会自己转换，无需再通过urllib.parse.quote转换
            return self.client.presigned_get_object(self.bucket, self.parse_path(object_name))
        return self.client.presigned_get_object(self.bucket, urllib.parse.quote(self.parse_path(object_name)))

    def delete(self, object_name):
        path = self.parse_path(object_name)
        self.client.remove_object(self.bucket, path)

引入上述类和minio交互，示例：

import os
from minuio_case.minio_oss import MinioOss


MINIO_CONF = {
    'host': '127.0.0.1:9000', # minio地址
    'access_key': 'minioadmin', # 账号
    'secret_key': 'minioadmin', # 密码
    'bucket': 'buckets1', # 桶
    'model_dirs': r'xcj' # 桶的文件夹
}

# 文件地址
file_path = r'E:\workspace\project\tools\tools\pdf\明天.pdf'

# 打开文件
with open(file_path, 'rb') as file:
    # 读取文件流
    content = file.read()
    
# 获取文件大小
file_size = os.path.getsize(file_path)

# 根据配置字典连接minio
minio_vessel = MinioOss(**MINIO_CONF)

# 上传本地文件
minio_vessel.upload_file('folder/明天.pdf',r'E:\workspace\project\tools\tools\pdf\明天.pdf')

# 上传文件流
minio_vessel.upload('folder/明天.pdf',content,file_size)

# 获取预览地址
preview_url = minio_vessel.preview_url("folder/明天.pdf")

作者：张猛