from minio import Minio file_name = '3e09ca66d9444906935b0171e26891f1.mp4' file_path = r'E:\集成资料\视频素材' barrel = "testdata" def upload_file(): # 创建minio客户端 client = Minio(endpoint="xxx.xxx.xxx.xxx:xxxxx", access_key='xxxxx', secret_key='xxxxx', secure=False # 使用http ) # 创建桶 client.make_bucket(bucket_name=barrel) # 删除桶 client.remove_bucket(barrel) # 获取桶列表 barrel_list = client.list_buckets() print(barrel_list) # 获取桶中的数据信息,不查子文件夹中的数据 bucket_objects = client.list_objects(barrel) for bucket_object in bucket_objects: print(bucket_object.object_name) # 列出名称以1-4开头的数据信息 bucket_objects = client.list_objects(barrel, prefix="1-4") for bucket_object in bucket_objects: print(bucket_object) # 递归遍历桶中的数据信息,读取子文件夹下的文件 bucket_objects = client.list_objects(barrel, recursive=True) for bucket_object in bucket_objects: print(bucket_object.object_name) # 递归查找以/data开头的数据信息 data = list() for root in ["/data1"]: bucket_objects = client.list_objects(barrel, prefix=root, recursive=True) for bucket_object in bucket_objects: data.append(bucket_object.object_name) # print(bucket_object.object_name) print(len(data)) # 递归查找以data1同一层级的数据信息 bucket_objects = client.list_objects(barrel, recursive=True, start_after="data1") for bucket_object in bucket_objects: print(bucket_object.object_name) # 上传文件, bucket_name: 桶名称, object_name:上传到桶中完整的文件路径, file_path:文件本地所在完整路径 result = client.fput_object(bucket_name=barrel, object_name="data1/" + file_name, file_path=file_path + "/" + file_name) print(result.object_name, result.bucket_name, result.etag) # 下载文件,bucket_name: 桶名称, object_name:被下载文件完整路径, file_path:保存到本地的完整路径 result = client.fget_object(bucket_name=barrel, object_name="data1/60719d5c50e833d4fa8af3b7412d40000a2.jpg", file_path=r"E:\集成资料\测试项目\minio\1.jpg") print(result.object_name, result.content_type, result.owner_name) # 判断桶是否存在 check_bucket = client.bucket_exists(barrel) if not check_bucket: # 不存在则创建桶 client.make_bucket(barrel) if __name__ == '__main__': upload_file()