当前位置：首页 > article >正文

【MinIO】Python 运用 MinIO 实现简易文件系统

article 2024/11/15 4:46:13

MinIO文件系统

1、Python & MinIO实现简易文件系统
2、环境准备
- 2.1 安装 MinIO 客户端
- - 2.1.1安装包方式
  - 2.1.2 MAC
- 2.2 启动 MinIO 客户端
- 2.3 Python 安装包
3、访问后台
- 生成秘钥
4、代码分析
- 4.1创建客户端
- 4.2 创建桶
- 4.3 上传文件
- 4.4 预览文件
- - 4.4.1 获取文件URL
  - 4.4.2 预览文件
5、完整代码实现
- 5.1 主体类
- 5.2 预览文件测试类

1、Python & MinIO实现简易文件系统

MinIO 是一个高性能的对象存储服务，今天学长给大家带来的是简易文件系统，教你如何使用 Python 操作 MinIO，实现文件的增加、删除、预览。

2、环境准备

2.1 安装 MinIO 客户端

2.1.1安装包方式

直接从以下官网下载安装包：

https://min.io/

2.1.2 MAC

直接输入以下命令：

brew install minio/stable/minio

2.2 启动 MinIO 客户端

提前创建目录，并赋予文件夹权限：

sudo chown -R username /usr/local/minio/data

启动MinIO，输入以下命令：

minio server /usr/local/minio/data

2.3 Python 安装包

输入以下命令：

pip install minio

3、访问后台

环境准备好之后，并启动后，可以直接访问后台：

http://127.0.0.1:53080/login

默认账号：minioadmin
默认密码：minioadmin

在这里插入图片描述

生成秘钥

在这里插入图片描述

4、代码分析

4.1创建客户端

   client = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=False)

4.2 创建桶

 def create_bucket(bucket_name):
      if not client.bucket_exists(bucket_name):
          client.make_bucket(bucket_name)
          print(f"Bucket {bucket_name} created successfully.")

在这里插入图片描述

4.3 上传文件

def put_obj_by_path(file_path,object_name):
    # 使用put_object方法上传文件（需要手动打开文件）
    try:
        with open(file_path, 'rb') as file_data:
            client.put_object(bucket_name, object_name, file_data, length=os.path.getsize(file_path))
            print(f"File {object_name} uploaded successfully.")
    except S3Error as err:
        print(err)

在这里插入图片描述

4.4 预览文件

4.4.1 获取文件URL

def get_file_url(bucket_name,destination_file):
    expires = timedelta(days=1)
    try:
        presigned_url = client.presigned_get_object(bucket_name, destination_file, expires=expires)
        print(f"Presigned URL for {destination_file}: {presigned_url}")
    except Exception as err:
        print(f"Failed to generate presigned URL: {err}")

在这里插入图片描述

4.4.2 预览文件

from flask import Flask, render_template_string

app = Flask(__name__)


@app.route('/')
def index():
    # 在这里你可以通过某种方式获取或传递image_url
    # 这里我们假设你已经生成了image_url
    template = '''  
    <!doctype html>  
    <html>  
        <head><title>Image Preview</title></head>  
        <body>  
            <h1>Image Preview</h1>  
            <img src="{{ image_url }}" alt="Image Preview">  
        </body>  
    </html>  
    '''
    return render_template_string(template, image_url='xxxxxx')


if __name__ == '__main__':
    # 启动Flask应用
    app.run(debug=True, host='0.0.0.0', port=5001)

在这里插入图片描述

5、完整代码实现

5.1 主体类

import os
from datetime import timedelta

from minio import Minio
from minio.error import S3Error

# MinIO服务地址、端口、访问密钥和秘密密钥
endpoint = "127.0.0.1:9000"
access_key = "4y2D7sVaiBtkMZEjfZP2"
secret_key = "PDth008ESsqLhRT5wR8dVQpdvEHbC0QvepF4YKM3"


def create_bucket(bucket_name):
    if not client.bucket_exists(bucket_name):
        client.make_bucket(bucket_name)
        print(f"Bucket {bucket_name} created successfully.")

def put_obj_by_path(file_path,object_name):
    # 使用put_object方法上传文件（需要手动打开文件）
    try:
        with open(file_path, 'rb') as file_data:
            client.put_object(bucket_name, object_name, file_data, length=os.path.getsize(file_path))
            print(f"File {object_name} uploaded successfully.")
    except S3Error as err:
        print(err)

def fput_obj_by_path(file_path,object_name):
    # 或者使用fput_object方法上传文件（直接指定文件路径）
    try:
        client.fput_object(bucket_name, object_name, file_path, content_type='text/plain')
        print(f"File {object_name} uploaded with fput_object.")
    except S3Error as err:
        print(err)

def download_file(bucket_name,object_name,path):
    # 下载文件并保存到本地
    try:
        data = client.get_object(bucket_name, object_name)
        with open(path, "wb") as fp:
            for d in data.stream(1024):
                fp.write(d)
        print(f"File {object_name} downloaded successfully.")
    except S3Error as err:
        print(err)

def del_file(bucket_name,object_name):
    try:
        client.remove_object(bucket_name, object_name)
        print(f"File {object_name} deleted successfully.")
    except S3Error as err:
        print(err)

def get_file_url(bucket_name,destination_file):
    expires = timedelta(days=1)
    try:
        presigned_url = client.presigned_get_object(bucket_name, destination_file, expires=expires)
        print(f"Presigned URL for {destination_file}: {presigned_url}")
    except Exception as err:
        print(f"Failed to generate presigned URL: {err}")

# 创建一个MinIO客户端实例
try:
    client = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=False)
    bucket_name = 'mybucket'
    create_bucket(bucket_name)
    # put_obj_by_path('./resources/i白工具库.jpg','test01')
    # fput_obj_by_path('./resources/i白工具库.jpg','test02')
    # down_load_path = './resources/downFile.jpg'
    # download_file(bucket_name,'test01',down_load_path)
    # del_file(bucket_name,'test02')
    get_file_url(bucket_name,'test01')
except S3Error as err:
    print(err)

5.2 预览文件测试类

from flask import Flask, render_template_string

app = Flask(__name__)


@app.route('/')
def index():
    # 在这里你可以通过某种方式获取或传递image_url
    # 这里我们假设你已经生成了image_url
    template = '''  
    <!doctype html>  
    <html>  
        <head><title>Image Preview</title></head>  
        <body>  
            <h1>Image Preview</h1>  
            <img src="{{ image_url }}" alt="Image Preview">  
        </body>  
    </html>  
    '''
    return render_template_string(template, image_url='http://127.0.0.1:9000/mybucket/test01?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=4y2D7sVaiBtkMZEjfZP2%2F20241016%2Fus-east-1%2Fs3%2Faws4_request&X-Amz-Date=20241016T145610Z&X-Amz-Expires=86400&X-Amz-SignedHeaders=host&X-Amz-Signature=06612dc63d674b83fc1b32fc2c00c67ec46ee47f7dea1d95c79eb4ea5ef188f6')


if __name__ == '__main__':
    # 启动Flask应用
    app.run(debug=True, host='0.0.0.0', port=5001)