39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
from minio import Minio
|
|
from minio.error import S3Error
|
|
import io
|
|
|
|
# 1. 创建客户端
|
|
client = Minio(
|
|
'localhost:9100',
|
|
access_key='minioadmin',
|
|
secret_key='minioadmin',
|
|
secure=False, # 本地测试用 HTTP;生产环境请用 True
|
|
)
|
|
|
|
bucket_name = 'test-bucket'
|
|
object_name = 'test-object'
|
|
data = b'Hello MinIO'
|
|
|
|
try:
|
|
# 2. 桶不存在就创建(关键修改)
|
|
if not client.bucket_exists(bucket_name):
|
|
client.make_bucket(bucket_name)
|
|
print(f"已创建桶: {bucket_name}")
|
|
|
|
# 3. 上传对象,length 用 len(data) 动态计算
|
|
client.put_object(
|
|
bucket_name,
|
|
object_name,
|
|
io.BytesIO(data),
|
|
length=len(data),
|
|
)
|
|
print(f"MinIO 连接成功,对象上传完成: {bucket_name}/{object_name}")
|
|
|
|
# 4. (可选)验证一下确实传上去了
|
|
stat = client.stat_object(bucket_name, object_name)
|
|
print(f"验证: 对象大小 = {stat.size} 字节")
|
|
|
|
except S3Error as e:
|
|
print(f"S3 错误: code={e.code}, message={e.message}")
|
|
except Exception as e:
|
|
print(f"其它错误: {type(e).__name__}: {e}") |