代码文件示例
from qingstor.sdk.service.qingstor import QingStor from qingstor.sdk.config import Config BUCKET_NAME = 'bucket_name' ZONE_KEY = 'zone_key' # 如果使用青云官网存储 # config = Config('ACCESS_KEY_ID_EXAMPLE', 'SECRET_ACCESS_KEY_EXAMPLE') # qingstor = QingStor(config) # 使用本地存储 配置文件单独配置 config = Config().load_config_from_filepath('./config.yaml') qingstor = QingStor(config) # List all buckets. output = qingstor.list_buckets() print(output.status_code) # 打印所有object. bucket = qingstor.Bucket(bucket_name=BUCKET_NAME, zone=ZONE_KEY) output = bucket.list_objects() print(output.status_code) for it in output.items(): print(it) for it in output['keys']: print(it['key']) # 获取 并保存到本地 object_key = "object_key" local_path = "./02.ipynb" resp = bucket.get_object(object_key=object_key) print(resp.status_code) print(resp.headers) print(resp.content) with open(local_path, 'wb') as f: f.write(resp.content) # 创建或者更新 with open(local_path, 'rb') as f: file_data = f.read() try: resp = bucket.put_object(object_key=object_key, body=file_data) if resp.status_code == 200: print(f"上传成功，保存为 {object_key}") print(f"ETag: {resp.headers.get('ETag')}") else: print(f"上传失败，状态码：{resp.status_code}") print(f"错误信息：{resp.content}") except FileNotFoundError: print(f"{local_path} 不存在") except Exception as e: print(e) # 删除 try: resp = bucket.delete_object(object_key=object_key) if resp.status_code == 204: print("删除成功") else: print(f"删除失败 {resp.status_code}") except Exception as e: print(e)config.yaml 示例
access_key_id: 'access_key_id' secret_access_key: 'secret_access_key' host: 'www.larance.xyz' port: 80 protocol: 'http' connect_retries: 3 Log_level: 'INFO'
