Simple Minio Python3 CLI Wrapper

minio_wrapper.py

#! /usr/bin/env python3

import os
import argparse
from minio import Minio
from minio.error import ResponseError


def download_file(session, bucket, minio_path, local_path):
    """
    Download an object by streaming to the FS with get_object
    """
    try:
        data = session.get_object(bucket, minio_path)
        with open(local_path, 'wb') as file_data:
            for d in data.stream(32*1024):
                file_data.write(d)

    except ResponseError as err:
        print(err)


def upload_file(session, bucket, minio_path, local_path):
    """
    Upload a local object using put_object
    """
    try:
        with open(local_path, 'rb') as file_data:
            size = os.stat(local_path).st_size
            session.put_object(bucket, minio_path, file_data, size)

    except ResponseError as err:
        print(err)


parser = argparse.ArgumentParser(description="A simple wrapper for downloading and uploading files to Minio")
parser.add_argument("-up", "--upload", action='store_true',
                    help="Run in upload mode.")
parser.add_argument("-dl", "--download", action='store_true',
                    help="Run in download mode.")

parser.add_argument("-url", "--minio_url",  metavar='URL',
                    help="Set the remote Minio URL to use")
parser.add_argument("-usr", "--username", metavar='ACCESS_KEY',
                    help="Set the Minio username")
parser.add_argument("-pwd", "--password", metavar='ACCESS_PASSWORD',
                    help="Set the Minio password")

parser.add_argument("-b", "--bucket",  metavar='INT',
                    help="Select the Minio bucket to upload/dowload from.")
parser.add_argument("-f", "--file", metavar='/PATH/TO/FILE',
                    help="Select the file to be uploaded/downloaded.")
parser.add_argument("-p", "--path",  metavar='INT',
                    help="Select the Minio path to upload/download from.")
args = parser.parse_args()


if args.upload or args.download is True:
    if args.bucket and args.username and args.password and args.file and args.path:

        client = Minio(args.minio_url,
                       access_key=args.username,
                       secret_key=args.password,
                       secure=True)

        if args.upload:
            upload_file(client, args.bucket, args.path, args.file)

        elif args.download:
            download_file(client, args.bucket, args.path, args.file)

    else:
        print('Not Enough Args! Read --help for more info!')
else:
    print('Not Enough Args! Read --help for more info!')