代码收藏家 【MinIO】Python 运用 MinIO 实现简易文件系统
MinIO文件系统
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)
作者：山河不见老
