Python如何操作MinIO存储？minio-py客户端

minio-py

minio-py

pip install minio

Minio

from minio import Minio
from minio.error import S3Error
import os

# 替换为你的MinIO服务器地址和凭证
MINIO_ENDPOINT = "127.0.0.1:9000" # 或者你的域名
MINIO_ACCESS_KEY = "minioadmin"
MINIO_SECRET_KEY = "minioadmin"
MINIO_SECURE = False # 如果是HTTPS，设为True

def run_minio_operations():
    try:
        # 1. 初始化MinIO客户端
        client = Minio(
            MINIO_ENDPOINT,
            access_key=MINIO_ACCESS_KEY,
            secret_key=MINIO_SECRET_KEY,
            secure=MINIO_SECURE
        )

        bucket_name = "my-test-bucket"
        object_name = "my-document.txt"
        file_path = "local_document.txt"
        download_path = "downloaded_document.txt"

        # 创建一个本地文件用于上传
        with open(file_path, "w") as f:
            f.write("Hello, MinIO! This is a test file.")
        print(f"本地文件 '{file_path}' 创建成功。")

        # 2. 检查并创建桶（如果不存在）
        if not client.bucket_exists(bucket_name):
            client.make_bucket(bucket_name)
            print(f"桶 '{bucket_name}' 创建成功。")
        else:
            print(f"桶 '{bucket_name}' 已存在。")

        # 3. 上传文件
        # fput_object 适合大文件，它会分块上传
        client.fput_object(
            bucket_name,
            object_name,
            file_path,
            content_type="text/plain" # 设定内容类型，浏览器访问时会用到
        )
        print(f"文件 '{object_name}' 已成功上传到桶 '{bucket_name}'。")

        # 4. 下载文件
        # fget_object 同样适合大文件
        client.fget_object(
            bucket_name,
            object_name,
            download_path
        )
        print(f"文件 '{object_name}' 已成功下载到 '{download_path}'。")

        # 5. 列出桶中的对象
        print(f"列出桶 '{bucket_name}' 中的对象：")
        objects = client.list_objects(bucket_name, recursive=True)
        for obj in objects:
            print(f"  - {obj.object_name} (大小: {obj.size} 字节)")

        # 6. 获取对象信息
        stat = client.stat_object(bucket_name, object_name)
        print(f"对象 '{object_name}' 的信息：Etag={stat.etag}, Size={stat.size}")

        # 7. 删除文件
        client.remove_object(bucket_name, object_name)
        print(f"文件 '{object_name}' 已从桶 '{bucket_name}' 中删除。")

        # 清理本地文件
        os.remove(file_path)
        os.remove(download_path)
        print("本地临时文件已清理。")

    except S3Error as e:
        print(f"MinIO操作发生S3错误: {e}")
    except Exception as e:
        print(f"发生未知错误: {e}")

if __name__ == "__main__":
    run_minio_operations()

minio-py

minio-py

minio-py

minio.error.S3Error

from minio import Minio
from minio.error import S3Error, ServerError, BucketAlreadyOwnedByYou, NoSuchBucket, NoSuchKey

# ... (MinIO客户端初始化代码同上)

def safe_upload_file(client, bucket_name, object_name, file_path):
    try:
        if not client.bucket_exists(bucket_name):
            print(f"桶 '{bucket_name}' 不存在，尝试创建...")
            client.make_bucket(bucket_name)
            print(f"桶 '{bucket_name}' 创建成功。")

        client.fput_object(bucket_name, object_name, file_path)
        print(f"文件 '{object_name}' 上传成功。")
    except BucketAlreadyOwnedByYou:
        # 这个错误通常发生在尝试创建已经属于你的桶时，可以忽略
        print(f"桶 '{bucket_name}' 已经存在，无需重复创建。")
    except S3Error as e:
        # 捕获所有S3相关的错误
        if e.code == "AccessDenied":
            print(f"权限不足：无法访问桶 '{bucket_name}' 或上传文件。请检查你的Access Key和Secret Key的权限。")
        elif e.code == "NoSuchBucket":
            print(f"桶不存在：'{bucket_name}'。请确认桶名是否正确或已创建。")
        elif e.code == "NoSuchKey":
            print(f"对象不存在：'{object_name}'。这通常在下载或删除不存在的对象时发生。")
        elif e.code == "InvalidAccessKeyId" or e.code == "SignatureDoesNotMatch":
            print(f"认证失败：Access Key或Secret Key不正确。请检查你的凭证。")
        else:
            print(f"MinIO操作发生S3错误: {e.code} - {e.message}")
    except ServerError as e:
        # 服务器端错误，例如MinIO服务不可用
        print(f"MinIO服务器错误: {e.message}。请检查MinIO服务是否正常运行。")
    except ConnectionRefusedError:
        print(f"连接被拒绝：无法连接到MinIO服务器 '{MINIO_ENDPOINT}'。请检查IP地址、端口和网络防火墙。")
    except Exception as e:
        # 捕获其他所有未预料的错误
        print(f"发生未知错误: {e}")

# 调用示例
# client = Minio(...)
# safe_upload_file(client, "non-existent-bucket", "test.txt", "some_local_file.txt")

AccessDenied

NoSuchBucket