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)