一、minio库核心概述
minio是一款用于访问MinIO对象存储服务的Python客户端库，能实现存储桶的创建、删除，以及对象的上传、下载、删除等操作。其工作原理是通过与MinIO服务端的API接口进行交互，遵循S3协议规范，可轻松对接私有部署的MinIO服务或兼容S3协议的云存储服务。该库优点是轻量高效、API简洁易懂、兼容性强；缺点是对复杂的分布式存储集群管理支持有限，需依赖服务端配置。minio库采用Apache License 2.0开源协议，允许商业和非商业自由使用与修改。
二、minio库安装方法
在使用minio库之前，我们需要先确保本地已经安装了Python环境（建议Python 3.6及以上版本），然后通过pip包管理工具即可完成安装，具体命令如下：
pip install minio执行上述命令后，pip会自动从PyPI下载并安装最新版本的minio库及其依赖项。安装完成后，我们可以在Python脚本中通过import minio来验证是否安装成功，若没有报错，则说明安装完成。
三、minio库核心操作实战
3.1 初始化MinIO客户端连接
要使用minio库操作MinIO对象存储，第一步是创建客户端连接对象，需要传入服务端的地址、Access Key、Secret Key以及是否启用HTTPS等参数。
代码示例
from minio import Minio
from minio.error import S3Error
def create_minio_client():
# 配置MinIO服务端信息
minio_server = "localhost:9000" # MinIO服务地址，端口默认9000
access_key = "your-access-key" # 访问密钥，对应服务端配置
secret_key = "your-secret-key" # 秘密密钥，对应服务端配置
# 初始化客户端，secure=False表示不使用HTTPS，生产环境建议开启
client = Minio(
minio_server,
access_key=access_key,
secret_key=secret_key,
secure=False
)
return client
if __name__ == "__main__":
try:
client = create_minio_client()
print("MinIO客户端连接成功！")
except S3Error as e:
print(f"客户端连接失败: {e}")代码说明
- 首先导入
Minio类和异常处理类S3Error，S3Error用于捕获MinIO操作过程中可能出现的各类错误。 create_minio_client函数中，填写MinIO服务端的实际地址、Access Key和Secret Key，这些参数需要与MinIO服务端的配置保持一致。secure=False表示使用HTTP协议连接，若服务端配置了HTTPS，则需要将其改为True。- 通过
if __name__ == "__main__"主程序入口，调用函数创建客户端并测试连接，成功则打印提示信息，失败则捕获异常并输出错误原因。
3.2 存储桶（Bucket）操作
存储桶是MinIO中用于存放对象的容器，类似于文件系统中的文件夹，但级别更高。minio库提供了创建、查询、删除存储桶的完整方法。
3.2.1 创建存储桶
当我们需要存储对象时，首先要创建一个对应的存储桶，创建前可以先判断该存储桶是否已经存在，避免重复创建。
代码示例
from minio import Minio
from minio.error import S3Error
def create_bucket(client, bucket_name):
# 检查存储桶是否存在
if not client.bucket_exists(bucket_name):
# 创建存储桶
client.make_bucket(bucket_name)
print(f"存储桶 {bucket_name} 创建成功！")
else:
print(f"存储桶 {bucket_name} 已存在，无需重复创建。")
if __name__ == "__main__":
try:
# 初始化客户端
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
# 定义要创建的存储桶名称
bucket_name = "my-first-bucket"
# 调用创建函数
create_bucket(client, bucket_name)
except S3Error as e:
print(f"存储桶操作失败: {e}")代码说明
bucket_exists方法接收存储桶名称作为参数，返回布尔值，True表示存在，False表示不存在。make_bucket方法用于创建新的存储桶，参数为存储桶名称，需要注意存储桶名称的命名规范：只能包含小写字母、数字、连字符（-），且不能以连字符开头或结尾，长度在3-63个字符之间。
3.2.2 列出所有存储桶
在实际开发中，我们可能需要查看当前MinIO服务中所有的存储桶信息，minio库提供了list_buckets方法来实现这个功能。
代码示例
from minio import Minio
from minio.error import S3Error
def list_all_buckets(client):
# 列出所有存储桶
buckets = client.list_buckets()
print("当前MinIO服务中的存储桶列表：")
for bucket in buckets:
print(f"存储桶名称: {bucket.name}, 创建时间: {bucket.creation_date}")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
list_all_buckets(client)
except S3Error as e:
print(f"获取存储桶列表失败: {e}")代码说明
list_buckets方法返回一个存储桶对象的列表，每个存储桶对象包含name（存储桶名称）和creation_date（创建时间）两个核心属性。- 通过循环遍历列表，我们可以逐个打印出所有存储桶的详细信息，方便进行管理和查看。
3.2.3 删除存储桶
对于不再需要的存储桶，我们可以使用remove_bucket方法将其删除，需要注意的是，删除存储桶前必须确保该存储桶内没有任何对象，否则删除操作会失败。
代码示例
from minio import Minio
from minio.error import S3Error
def delete_bucket(client, bucket_name):
# 先检查存储桶是否存在
if client.bucket_exists(bucket_name):
# 尝试删除存储桶
client.remove_bucket(bucket_name)
print(f"存储桶 {bucket_name} 删除成功！")
else:
print(f"存储桶 {bucket_name} 不存在，无需删除。")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
bucket_name = "my-first-bucket"
delete_bucket(client, bucket_name)
except S3Error as e:
print(f"删除存储桶失败: {e}")代码说明
remove_bucket方法接收存储桶名称作为参数，执行删除操作。- 如果存储桶内存在对象，调用该方法会抛出
S3Error异常，提示“BucketNotEmpty”，因此在删除前可以先列出桶内对象并删除，再执行桶删除操作。
3.3 对象（Object）操作
对象是MinIO存储的基本单元，对应文件系统中的文件，minio库支持对象的上传、下载、删除、列出等核心操作，是日常开发中最常用的功能模块。
3.3.1 上传本地文件到存储桶
将本地文件上传到MinIO存储桶是最基础的需求之一，minio库提供了fput_object方法，支持本地文件的直接上传。
代码示例
from minio import Minio
from minio.error import S3Error
def upload_file_to_bucket(client, bucket_name, local_file_path, remote_file_name):
# 检查存储桶是否存在，不存在则创建
if not client.bucket_exists(bucket_name):
client.make_bucket(bucket_name)
print(f"存储桶 {bucket_name} 不存在，已自动创建。")
# 上传文件
client.fput_object(
bucket_name=bucket_name,
object_name=remote_file_name,
file_path=local_file_path
)
print(f"文件 {local_file_path} 已成功上传到存储桶 {bucket_name}，远程文件名: {remote_file_name}")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
bucket_name = "my-file-bucket"
local_file = "./test.txt" # 本地待上传的文件路径
remote_name = "uploaded_test.txt" # 上传到存储桶后的文件名
upload_file_to_bucket(client, bucket_name, local_file, remote_name)
except S3Error as e:
print(f"文件上传失败: {e}")代码说明
fput_object方法有三个核心参数：bucket_name（目标存储桶名称）、object_name（上传后的对象名称）、file_path（本地文件的路径）。- 该方法会自动读取本地文件并上传到指定存储桶，上传完成后，我们可以通过MinIO的控制台或者其他工具查看上传的对象。
- 如果需要上传大文件，可以考虑使用分片上传功能，minio库也提供了对应的
multipart_upload相关方法。
3.3.2 从存储桶下载文件到本地
与上传操作相对应，fget_object方法可以将存储桶中的对象下载到本地指定路径。
代码示例
from minio import Minio
from minio.error import S3Error
def download_file_from_bucket(client, bucket_name, remote_file_name, local_file_path):
# 检查存储桶是否存在
if not client.bucket_exists(bucket_name):
print(f"存储桶 {bucket_name} 不存在，无法下载文件。")
return
# 下载文件
client.fget_object(
bucket_name=bucket_name,
object_name=remote_file_name,
file_path=local_file_path
)
print(f"文件 {remote_file_name} 已从存储桶 {bucket_name} 下载到本地 {local_file_path}")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
bucket_name = "my-file-bucket"
remote_name = "uploaded_test.txt"
local_file = "./downloaded_test.txt"
download_file_from_bucket(client, bucket_name, remote_name, local_file)
except S3Error as e:
print(f"文件下载失败: {e}")代码说明
fget_object方法的参数与fput_object类似，object_name为存储桶中的对象名称，file_path为本地保存的路径。- 如果指定的本地路径已存在同名文件，该方法会覆盖原有文件，因此在使用时需要注意文件的唯一性。
3.3.3 列出存储桶中的所有对象
当需要查看某个存储桶内的所有对象时，可以使用list_objects方法，该方法支持分页查询和前缀过滤。
代码示例
from minio import Minio
from minio.error import S3Error
def list_objects_in_bucket(client, bucket_name, prefix=None):
if not client.bucket_exists(bucket_name):
print(f"存储桶 {bucket_name} 不存在。")
return
# 列出桶内对象，prefix用于过滤对象名称前缀
objects = client.list_objects(bucket_name, prefix=prefix, recursive=True)
print(f"存储桶 {bucket_name} 中的对象列表：")
for obj in objects:
print(f"对象名称: {obj.object_name}, 大小: {obj.size} bytes, 最后修改时间: {obj.last_modified}")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
bucket_name = "my-file-bucket"
# 过滤前缀为"uploaded"的对象，若不指定前缀则传入None
list_objects_in_bucket(client, bucket_name, prefix="uploaded")
except S3Error as e:
print(f"列出对象失败: {e}")代码说明
list_objects方法的prefix参数用于过滤对象名称，例如传入prefix="uploaded"，则只会列出名称以“uploaded”开头的对象。recursive=True表示递归列出所有对象，包括子目录下的对象（MinIO中没有真正的目录，通过对象名称中的/模拟目录结构）。- 每个对象包含
object_name（对象名称）、size（大小）、last_modified（最后修改时间）等属性，方便获取对象的详细信息。
3.3.4 删除存储桶中的对象
对于不需要的对象，可以使用remove_object方法进行删除，同时minio库还支持remove_objects方法批量删除多个对象。
代码示例（单个删除）
from minio import Minio
from minio.error import S3Error
def delete_single_object(client, bucket_name, object_name):
if not client.bucket_exists(bucket_name):
print(f"存储桶 {bucket_name} 不存在。")
return
# 删除单个对象
client.remove_object(bucket_name, object_name)
print(f"对象 {object_name} 已从存储桶 {bucket_name} 中删除。")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
bucket_name = "my-file-bucket"
object_name = "uploaded_test.txt"
delete_single_object(client, bucket_name, object_name)
except S3Error as e:
print(f"删除对象失败: {e}")代码示例（批量删除）
from minio import Minio
from minio.error import S3Error
def delete_multiple_objects(client, bucket_name, object_names):
if not client.bucket_exists(bucket_name):
print(f"存储桶 {bucket_name} 不存在。")
return
# 构建待删除对象的迭代器
delete_objects = [{"name": name} for name in object_names]
errors = client.remove_objects(bucket_name, delete_objects)
# 处理删除过程中的错误
for error in errors:
print(f"删除对象 {error._object_name} 失败: {error}")
print("批量删除操作执行完成。")
if __name__ == "__main__":
try:
client = Minio(
"localhost:9000",
access_key="your-access-key",
secret_key="your-secret-key",
secure=False
)
bucket_name = "my-file-bucket"
objects_to_delete = ["uploaded_test.txt", "test_image.jpg"]
delete_multiple_objects(client, bucket_name, objects_to_delete)
except S3Error as e:
print(f"批量删除失败: {e}")代码说明
- 单个删除使用
remove_object方法，参数为存储桶名称和对象名称。 - 批量删除使用
remove_objects方法，需要传入一个包含对象名称的字典列表，该方法会返回一个错误迭代器，用于捕获删除失败的对象信息。
四、minio库实际应用案例——文件备份工具
在实际开发中，我们可以利用minio库快速实现一个本地文件备份工具，自动将指定目录下的文件备份到MinIO存储桶中，下面是完整的案例代码。
4.1 需求分析
- 遍历本地指定目录下的所有文件；
- 将文件自动上传到MinIO存储桶，以文件的修改时间作为前缀，避免文件名冲突；
- 上传完成后，记录备份日志。
4.2 完整代码实现
import os
import time
from minio import Minio
from minio.error import S3Error
class MinioFileBackup:
def __init__(self, minio_server, access_key, secret_key, secure=False):
# 初始化MinIO客户端
self.client = Minio(
minio_server,
access_key=access_key,
secret_key=secret_key,
secure=secure
)
def create_bucket_if_not_exists(self, bucket_name):
"""如果存储桶不存在则创建"""
if not self.client.bucket_exists(bucket_name):
self.client.make_bucket(bucket_name)
print(f"存储桶 {bucket_name} 创建成功")
def backup_files(self, local_dir, bucket_name):
"""备份本地目录下的所有文件到MinIO存储桶"""
# 检查本地目录是否存在
if not os.path.isdir(local_dir):
print(f"本地目录 {local_dir} 不存在")
return
# 创建存储桶（如果不存在）
self.create_bucket_if_not_exists(bucket_name)
# 遍历本地目录
for root, dirs, files in os.walk(local_dir):
for file in files:
local_file_path = os.path.join(root, file)
# 获取文件的修改时间，格式化为YYYYMMDDHHMMSS
mtime = os.path.getmtime(local_file_path)
time_str = time.strftime("%Y%m%d%H%M%S", time.localtime(mtime))
# 构建远程对象名称，格式：时间前缀_原文件名
remote_file_name = f"{time_str}_{file}"
try:
# 上传文件
self.client.fput_object(
bucket_name=bucket_name,
object_name=remote_file_name,
file_path=local_file_path
)
# 记录备份日志
log_content = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 成功备份文件: {local_file_path} -> {bucket_name}/{remote_file_name}\n"
self.write_backup_log(log_content)
print(f"成功备份: {local_file_path}")
except S3Error as e:
error_log = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] 备份文件失败: {local_file_path}, 错误信息: {e}\n"
self.write_backup_log(error_log)
print(f"备份失败: {local_file_path}, 错误: {e}")
def write_backup_log(self, log_content):
"""写入备份日志到本地文件"""
with open("backup_log.txt", "a", encoding="utf-8") as f:
f.write(log_content)
if __name__ == "__main__":
# 配置MinIO连接信息
MINIO_SERVER = "localhost:9000"
ACCESS_KEY = "your-access-key"
SECRET_KEY = "your-secret-key"
# 配置备份参数
LOCAL_BACKUP_DIR = "./backup_source" # 本地待备份的目录
BUCKET_NAME = "my-backup-bucket" # 目标存储桶名称
# 创建备份工具实例并执行备份
backup_tool = MinioFileBackup(MINIO_SERVER, ACCESS_KEY, SECRET_KEY, secure=False)
backup_tool.backup_files(LOCAL_BACKUP_DIR, BUCKET_NAME)4.3 代码说明
- 定义
MinioFileBackup类，封装MinIO客户端初始化、存储桶创建、文件备份和日志记录等功能； __init__方法接收MinIO服务端信息，初始化客户端；create_bucket_if_not_exists方法用于检查并创建存储桶；backup_files方法是核心功能，通过os.walk遍历本地目录下的所有文件，获取每个文件的修改时间并作为前缀，避免上传到MinIO后文件名冲突；- 上传文件时捕获
S3Error异常，成功和失败的信息都会写入本地的backup_log.txt日志文件； write_backup_log方法负责将日志内容追加写入日志文件，方便后续查看备份记录。
五、minio库相关资源地址
- PyPI地址：https://pypi.org/project/minio
- Github地址：https://github.com/minio/minio-py
- 官方文档地址：https://min.io/docs/minio/linux/developers/python/API.html
关注我，每天分享一个实用的Python自动化工具。
