Untitled

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

class dotdict(dict):
    """dot.notation access to dictionary attributes"""
    __getattr__ = dict.get
    __setattr__ = dict.__setitem__
    __delattr__ = dict.__delitem__

def main():
    client = Minio(
        endpoint="localhost:9001",
        access_key="sH8YvadobA6jUOLf",
        secret_key="A6Jlc6msYpbsdrhgyNNP4UKMVAVatkdO",
        secure=False
    )

    bucket_name = "vehicle-audit-video"
    video_path = os.path.abspath("test.mp4")
    object_name = "test.mp4"

    found = client.bucket_exists(bucket_name)
    if not found:
        client.make_bucket(bucket_name)
    else:
        print(f"Bucket {bucket_name} already exists")

    chunksize = 1024*1024*10

    upload_id = client._create_multipart_upload(bucket_name, object_name, headers={
        'Content-Type': 'video/mp4'
    })

    parts = []
    part_number = 1
    with open(video_path, "rb") as f:       
        while True:
            data = f.read(chunksize)
            if not data:
                break

            e_tag = client._upload_part(
                bucket_name,
                object_name,
                data,
                {
                    'Content-Type': 'video/mp4'
                },
                upload_id,
                part_number,
            )

            parts.append(dotdict({
                'part_number': part_number,
                'etag': e_tag
            }))

            part_number += 1

    try:
        client._complete_multipart_upload(
        bucket_name,
        object_name,
        upload_id,
        parts
    )
    except Exception as err:
        print(err)

if __name__ == "__main__":
    try:
        main()
    except S3Error as exc:
        print("error occurred.", exc)