封装Python类管理MinIO-阿南达文事网

封装Python类管理MinIO

封装一个用于管理 MinIO 上传、删除、修改等操作的类。这个类将提供类似于你当前代码的文件上传功能，但文件会上传到 MinIO 而不是本地文件系统。

安装 MinIO SDK

首先，确保你安装了 minio Python 客户端：

代码语言：javascript代码运行次数：0运行复制

pip install minio

1、Python操作MinIO

封装 MinIO 文件操作类

代码语言：javascript代码运行次数：0运行复制

from pathlib import Path

from minio import Minio
from minio.error import S3Error
import os

from xxx.settings import AWS_S3_ENDPOINT_IP, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, logger, STATIC_ROOT


class MinioManager:
    def __init__(self, endpoint, access_key, secret_key, secure=False):
        """
        初始化Minio客户端
        :param endpoint: Minio服务的URL
        :param access_key: Minio访问密钥
        :param secret_key: Minio密钥
        :param secure: 是否使用https
        """
        self.client = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure)

    def upload_file(self, bucket_name, file_path_or_obj, object_name=None):
        """
        上传文件到Minio
        :param bucket_name: 存储桶名称
        :param file_path_or_obj: 本地文件路径或文件对象
        :param object_name: Minio上存储的对象名称，默认是文件名
        :return: 上传结果
        """
        try:
            # 确保存储桶存在
            if not self.client.bucket_exists(bucket_name):
                self.client.make_bucket(bucket_name)

            # 如果输入的是文件路径
            if isinstance(file_path_or_obj, str):
                file_path = file_path_or_obj
                if object_name is None:
                    object_name = os.path.basename(file_path)
                self.client.fput_object(bucket_name, object_name, file_path)
            # 如果输入的是文件对象
            elif hasattr(file_path_or_obj, 'read'):
                file_obj = file_path_or_obj
                if object_name is None:
                    raise ValueError("If file is an object, object_name must be provided.")
                self.client.put_object(bucket_name, object_name, file_obj, file_obj.length)
            else:
                raise ValueError("Invalid file input type.")
            logger.info(f"File {object_name} uploaded successfully to {bucket_name}")
            return True
        except S3Error as e:
            logger.error(f"Error uploading file: {e}")
            return False

    def download_file(self, bucket_name, object_name, file_path):
        """
        下载文件从Minio
        :param bucket_name: 存储桶名称
        :param object_name: Minio上存储的对象名称
        :param file_path: 本地保存的路径
        :return: 下载结果
        """
        try:
            self.client.fget_object(bucket_name, object_name, file_path)
            logger.info(f"File {object_name} downloaded successfully to {file_path}")
            return True
        except S3Error as e:
            logger.error(f"Error downloading file: {e}")
            return False

    def delete_file(self, bucket_name, object_name):
        """
        删除文件
        :param bucket_name: 存储桶名称
        :param object_name: Minio上存储的对象名称
        :return: 删除结果
        """
        try:
            self.client.remove_object(bucket_name, object_name)
            logger.info(f"File {object_name} deleted successfully from {bucket_name}")
            return True
        except S3Error as e:
            logger.error(f"Error deleting file: {e}")
            return False

    def upload_directory(self, bucket_name, directory_path, new_directory=""):
        """
        上传目录下的所有文件到Minio，并保留原有的目录结构。
        :param bucket_name: 存储桶名称
        :param directory_path: 本地目录路径
        :param new_directory: MinIO中的目标目录路径
        :return: 上传结果
        """
        try:
            # 确保存储桶存在
            if not self.client.bucket_exists(bucket_name):
                self.client.make_bucket(bucket_name)

            # 遍历目录，上传文件
            for root, dirs, files in os.walk(directory_path):
                for file in files:
                    file_path = os.path.join(root, file)

                    # 计算相对路径并转换为 POSIX 格式
                    object_name = Path(file_path).relative_to(Path(directory_path)).as_posix()

                    # 如果指定了新的目录路径，将其加到目标路径上
                    if new_directory:
                        object_name = Path(new_directory, object_name).as_posix()

                    # 上传文件，保持目录结构
                    self.client.fput_object(bucket_name, object_name, file_path)
                    logger.info(f"Uploaded file {file_path} as {object_name} to {bucket_name}")

            logger.info(f"All files from directory {directory_path} uploaded successfully to {bucket_name}")
            return True
        except S3Error as e:
            logger.error(f"Error uploading directory: {e}")
            return False

    # 删除指定桶中的所有文件（模拟删除目录）
    def delete_directory(self, bucket_name, directory_name):
        try:
            # 列出桶中所有对象，过滤出目录下的对象
            objects = self.client.list_objects(bucket_name, prefix=directory_name, recursive=True)
            for obj in objects:
                # 删除每个对象
                self.client.remove_object(bucket_name, obj.object_name)
                logger.info(f"删除对象：{obj.object_name}")

            logger.info(f"目录 {directory_name} 下的所有对象已删除。")
            return True
        except S3Error as err:
            logger.error(f"删除目录发生错误: {err}")
            return False

    def modify_file(self, bucket_name, object_name, new_file_path_or_obj):
        """
        修改Minio上的文件，实际上是删除旧文件然后上传新文件
        :param bucket_name: 存储桶名称
        :param object_name: Minio上存储的对象名称
        :param new_file_path_or_obj: 新文件的路径或文件对象
        :return: 修改结果
        """
        try:
            # 删除旧文件
            self.delete_file(bucket_name, object_name)
            # 上传新文件
            return self.upload_file(bucket_name, new_file_path_or_obj, object_name)
        except S3Error as e:
            logger.error(f"Error modifying file: {e}")
            return False


# 初始化 MinIOStorage
# MinIO安装包专用桶
minio_client = MinioManager(
    endpoint=AWS_S3_ENDPOINT_IP,
    access_key=AWS_ACCESS_KEY_ID,
    secret_key=AWS_SECRET_ACCESS_KEY,
    secure=False
)

# 示例使用
if __name__ == "__main__":
    # minio_client = MinioManager(endpoint='minio.example', access_key='your-access-key',
    #                             secret_key='your-secret-key')

    # 上传单个文件
    # result = minio_client.upload_file('gr-test', 'E:\code\gr_backend\static\game\__INLINE__58.png', 'static/game/__INLINE__58.png')
    # print(result)
    #
    # # 下载文件
    # result = minio_client.download_file('mybucket', 'file.txt', 'path/to/local/save/file.txt')
    # print(result)
    #
    # # 删除文件
    # result = minio_client.delete_file('mybucket', 'file.txt')
    # print(result)

    # 上传目录
    image_path = os.path.join(STATIC_ROOT, 'perfetto', '2022-11-30')
    result = minio_client.upload_directory(bucket_name='perfetto', directory_path=image_path, new_directory='2022-11-30')
    print(result)

    # 删除目录
    # result = minio_client.delete_directory('perfetto', '')
    # print(result)

    # # 修改文件
    # result = minio_client.modify_file('mybucket', 'file.txt', 'path/to/new/local/file.txt')
    # print(result)

使用这个 MinIOStorage 类上传文件

你可以在 Django 中调用这个类来将文件上传到 MinIO。假设你已经将文件对象传递给 Django 的视图函数，下面是如何实现上传到 MinIO：

代码语言：javascript代码运行次数：0运行复制

# 上传目录到Minio, 并清理本地目录
minio_client.upload_directory(bucket_name='perfdog-image', directory_path=image_path, new_directory=file_key)
clear_dir(image_path)

2、Django中使用MinIO

首先在django settings.py中进行配置

代码语言：javascript代码运行次数：0运行复制

DEFAULT_FILE_STORAGE = 'storages.backends.s3boto3.S3Boto3Storage'

AWS_S3_ENDPOINT_URL = 'http://xxx:9000'  # MinIO 服务端地址
AWS_ACCESS_KEY_ID = 'minioadmin'  # 在 MinIO 中配置的访问密钥
AWS_SECRET_ACCESS_KEY = 'minioadmin'
AWS_STORAGE_BUCKET_NAME = 'test'  # 桶名
AWS_S3_REGION_NAME = 'us-east-1'  # 可自定义
AWS_S3_ADDRESSING_STYLE = "path"

写入文件（上传文件）

这是你提供的示例代码，上传文件到 MinIO。

代码语言：javascript代码运行次数：0运行复制

from django.core.files.storage import default_storage

def save_upload_image(fp, path):
    with default_storage.open(path, 'wb') as f:
        f.write(fp.read())

fp 是文件对象（例如，通过表单上传的文件），path 是存储路径。
default_storage.open(path, 'wb') 将文件内容写入 MinIO 中指定的路径。

删除文件

你可以使用 default_storage.delete() 来删除存储在 MinIO 中的文件。

代码语言：javascript代码运行次数：0运行复制

def delete_image(path):
    if default_storage.exists(path):
        default_storage.delete(path)

default_storage.exists(path) 用来检查文件是否存在。
default_storage.delete(path) 删除指定路径的文件。

修改文件

对于修改文件，通常是先删除旧文件，然后上传新文件。你可以结合上面的删除和写入操作来实现。

代码语言：javascript代码运行次数：0运行复制

def update_image(fp, path):
    if default_storage.exists(path):
        default_storage.delete(path)
    
    with default_storage.open(path, 'wb') as f:
        f.write(fp.read())

先使用 default_storage.exists(path) 检查文件是否存在。
使用 default_storage.delete(path) 删除旧文件。
使用 default_storage.open(path, 'wb') 写入新的文件。

下载文件

你可以使用 default_storage.open() 来读取存储在 MinIO 中的文件。

代码语言：javascript代码运行次数：0运行复制

def download_image(path):
    if default_storage.exists(path):
        with default_storage.open(path, 'rb') as f:
            return f.read()
    return None

default_storage.open(path, 'rb') 以二进制模式读取文件。
如果文件存在，则返回文件的内容；如果不存在，则返回 None。

配置nginx重定向，避免直接访问minIO服务器

代码语言：javascript代码运行次数：0运行复制

# 使用 proxy_pass 转发到 Minio 服务器
    location /static/perfdog_image/ {
        proxy_pass http://9.130.160.194:9000/perfdog-image/;
        proxy_set_header Host $host;  # 保证请求头中传递 Host 信息
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

封装Python类管理MinIO