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

在 Ubuntu 上安装 MinIO 并使用 Python 封装类操作对象存储

news 2025/10/18 6:30:57

MinIO 是一个高性能、兼容 Amazon S3 API 的对象存储服务，非常适合小型私有云或者开发测试环境。本文将详细介绍在 Ubuntu 上安装 MinIO，并用 Python 封装类实现文件上传、下载、删除和批量操作。

一、MinIO 安装与配置

1. 下载 MinIO 二进制

wget https://dl.min.io/server/minio/release/linux-amd64/minio -O /usr/local/bin/minio
sudo chmod +x /usr/local/bin/minio

确保 MinIO 可执行文件有执行权限。

2. 创建数据目录

sudo mkdir -p /data/minio 
sudo chown -R $USER:$USER /data/minio

3. 以系统服务方式启动 MinIO（可选）

创建 systemd 服务文件 /etc/systemd/system/minio.service：

[Unit]
Description=MinIO Object Storage
After=network.target[Service]
User=root
ExecStart=/usr/local/bin/minio server /data/minio --console-address ":9001"
Restart=always
LimitNOFILE=65536[Install]
WantedBy=multi-user.target

启用并启动：

sudo systemctl daemon-reload
sudo systemctl enable minio
sudo systemctl start minio
sudo systemctl status minio

注意：如果在 WSL 环境下，systemd 默认未启用，可直接用命令启动：
minio server /data/minio --console-address ":9001"

4. 测试 MinIO

访问浏览器：

Web 控制台: http://127.0.0.1:9001
API 端口: http://127.0.0.1:9000

默认用户名和密码可以在启动命令中通过环境变量指定：

export MINIO_ROOT_USER=admin
export MINIO_ROOT_PASSWORD=admin123456

二、Python 操作 MinIO 的封装类

为了方便操作 MinIO 对象存储，我们可以写一个 Python 封装类 MinioClient，支持：

文件上传/下载
字节数据上传/下载
删除文件
列举对象
批量上传/下载文件夹
生成临时 URL

1. 安装 Python SDK

pip install minio

2. 封装类代码

from minio import Minio
from minio.error import S3Error
from io import BytesIO
import osclass MinioClient:def __init__(self, endpoint, access_key, secret_key, bucket_name, secure=False):"""初始化 MinIO 客户端，并固定 bucket"""self.client = Minio(endpoint, access_key=access_key, secret_key=secret_key, secure=secure)self.bucket_name = bucket_name# 自动创建 bucketif not self.client.bucket_exists(bucket_name):self.client.make_bucket(bucket_name)def upload_file(self, object_name, file_path, content_type=None):try:self.client.fput_object(self.bucket_name, object_name, file_path, content_type=content_type)except S3Error as e:raise RuntimeError(f"上传文件 {file_path} 到 {self.bucket_name}/{object_name} 失败: {e}")def upload_bytes(self, object_name, data: bytes, content_type="application/octet-stream"):try:self.client.put_object(self.bucket_name, object_name, BytesIO(data), length=len(data), content_type=content_type)except S3Error as e:raise RuntimeError(f"上传字节数据到 {self.bucket_name}/{object_name} 失败: {e}")def download_file(self, object_name, file_path):try:self.client.fget_object(self.bucket_name, object_name, file_path)except S3Error as e:raise RuntimeError(f"下载 {self.bucket_name}/{object_name} 到 {file_path} 失败: {e}")def download_bytes(self, object_name) -> bytes:try:response = self.client.get_object(self.bucket_name, object_name)data = response.read()response.close()response.release_conn()return dataexcept S3Error as e:raise RuntimeError(f"下载 {self.bucket_name}/{object_name} 失败: {e}")def delete_object(self, object_name):try:self.client.remove_object(self.bucket_name, object_name)except S3Error as e:raise RuntimeError(f"删除 {self.bucket_name}/{object_name} 失败: {e}")def list_objects(self, prefix="", recursive=True):try:return [obj.object_name for obj in self.client.list_objects(self.bucket_name, prefix=prefix, recursive=recursive)]except S3Error as e:raise RuntimeError(f"列举 {self.bucket_name} 下对象失败: {e}")def presigned_get_url(self, object_name, expires=3600):try:return self.client.presigned_get_object(self.bucket_name, object_name, expires=expires)except S3Error as e:raise RuntimeError(f"生成下载链接失败: {e}")def presigned_put_url(self, object_name, expires=3600):try:return self.client.presigned_put_object(self.bucket_name, object_name, expires=expires)except S3Error as e:raise RuntimeError(f"生成上传链接失败: {e}")# ================= 批量操作文件夹 =================def upload_folder(self, local_folder, remote_folder=""):"""上传本地目录到 MinIO"""if not os.path.isdir(local_folder):raise ValueError(f"{local_folder} 不是目录")for root, _, files in os.walk(local_folder):for file in files:local_path = os.path.join(root, file)relative_path = os.path.relpath(local_path, local_folder)object_name = os.path.join(remote_folder, relative_path).replace("\\", "/")self.upload_file(object_name, local_path)def download_folder(self, remote_folder, local_folder):"""下载 MinIO 前缀到本地目录"""objects = self.list_objects(prefix=remote_folder, recursive=True)for obj_name in objects:relative_path = os.path.relpath(obj_name, remote_folder).replace("\\", "/")local_path = os.path.join(local_folder, relative_path)os.makedirs(os.path.dirname(local_path), exist_ok=True)self.download_file(obj_name, local_path)

3. 使用示例

client = MinioClient("127.0.0.1:9000", "admin", "admin123456", bucket_name="mybucket")# 上传单文件
client.upload_file("test.txt", "/tmp/test.txt")# 上传字节数据
client.upload_bytes("hello.txt", b"Hello MinIO!")# 下载文件
client.download_file("test.txt", "/tmp/test_download.txt")# 下载到内存
data = client.download_bytes("hello.txt")
print(data.decode())# 删除文件
client.delete_object("hello.txt")# 列举对象
print(client.list_objects())# 批量上传目录
client.upload_folder("/tmp/local_folder", "remote_folder")# 批量下载目录
client.download_folder("remote_folder", "/tmp/download_folder")# 临时下载 URL
url = client.presigned_get_url("test.txt")
print("Download URL:", url)