Untitled
unknown
python
7 months ago
1.8 kB
3
Indexable
Never
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)