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

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

news 2024/11/23 13:15:30

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_stringapp = Flask(__name__)@app.route('/')
def index():# 在这里你可以通过某种方式获取或传递image_url# 这里我们假设你已经生成了image_urltemplate = '''  <!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 timedeltafrom 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_stringapp = Flask(__name__)@app.route('/')
def index():# 在这里你可以通过某种方式获取或传递image_url# 这里我们假设你已经生成了image_urltemplate = '''  <!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)