【MinIO】Python 运用 MinIO 实现简易文件系统
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)