Minio helper functions

minio_helpers.py

from minio import Minio
from minio.error import S3Error

def make_bucket(endpoint, access_key, secret_key, bucket_name):
    client = Minio(endpoint=endpoint, access_key=access_key, secret_key=secret_key, secure=False)
    try:
        return client.make_bucket(bucket_name)
    except S3Error as s3e:
        return s3e

def get_bucket_names(endpoint, access_key, secret_key):
    client = Minio(endpoint=endpoint, access_key=access_key, secret_key=secret_key, secure=False)
    return [b.name for b in client.list_buckets()]

def upload_file(endpoint, access_key, secret_key, bucket, files=[]):
    client = Minio(endpoint=endpoint, access_key=access_key, secret_key=secret_key, secure=False)
    if type(files) == str:
        fn = files.split('/')[-1]
        res = client.fput_object(bucket, fn, files)
        return res
    if len(files) == 0:
        return None
    res = []
    for f in files:
        fn = f.split('/')[-1]
        res.append(client.fput_object(bucket, fn, f))
    return res

def list_objects(endpoint, access_key, secret_key, bucket):
    result = {}
    client = Minio(endpoint=endpoint, access_key=access_key, secret_key=secret_key, secure=False)
    for o in client.list_objects(bucket): # object_name, size, last_modified
        result[o.object_name] = {"size": o.size, "last_modified": str(o.last_modified), "owner": o.owner_name}
    return result