AI 助理
备案控制台
官方文档
输入文档关键字查找
首页

Python SDK示例程序

更新时间：
我的收藏

Python SDK 简易使用示例

本节以示例的方式展示如何使用SDK高级接口进行开发。用户在阅读本节后，可模仿示例，并参考高级接口一节进行开发。

其中，方括号内的参数用户应根据实际需求进行替换。

  • 创建 Vault
  1. from oas.oas_api import OASAPI
  2. from oas.ease.vault import Vault
  4. # 创建 OASAPI 对象
  5. api = OASAPI('[Server Host]', '[Access Key ID]', '[Access Key Secret]')
  7. # 创建 Vault
  8. vault = Vault.create_vault(api, '[Vault Name]')
  • 查找Vault
  1. # 创建 OASAPI 对象
  2. api = OASAPI('[Server Host]', '[Access Key ID]', '[Access Key Secret]')
  4. # 根据名称获取 Vault
  5. vault = Vault.get_vault_by_name(api, '[Vault Name]')
  7. # 根据 ID 获取 Vault
  8. vault = Vault.get_vault_by_id(api, '[Vault ID]')
  • 上传文件
  1. archive_id = vault.upload_archive('[File Path]')
  • 删除 Archive
  1. vault.delete_archive('[Archive ID]')
  • 续传 Multipart Upload 任务
  1. uploader = vault.recover_uploader('[Upload ID]')
  2. uploader.resume('[File Path]')
  • 获取 Archive 列表
  1. job = vault.retrieve_inventory()
  2. job.download_to_file('[File Path]')
  • 下载 Archive
  1. job = vault.retrieve_archive('[Archive ID]')
  2. job.download_to_file('[File Path]')
  • 从OSS上转储Object到OAS
  1. job = vault.pull_from_oss(conf.osshost, conf.bucket, conf.object, "test desc")
  • 从OAS上转储Archive到OSS
  1. job = vault.push_to_oss(archive_id, conf.osshost, conf.bucket, archive_id, "test desc")

Python SDK 完整使用演示代码

下面函数中test_single_archive_upload提供单一文件archive上传；

函数test_multi_upload() 使用sdk低级接口实现分段上传；

函数test_uploader()使用sdk高级接口实数据上传（当数据大于64MB时，会自动选择分段上传）；

函数test_vault_retrieve()实现vault信息查询；

函数test_download_archive(archive_id)实现archive下载；

函数test_delete_archive(archive_id)实现archive删除。

函数test_pull_from_oss（） 实现从OSS直接转储到OAS

函数test_push_to_oss() 实现从OAS直接转储到OSS

  1. import random
  2. import time 
  3. import logging
  5. import logging.handlers  
  6. from oas.oas_api import OASAPI
  7. from oas.ease.api import APIProxy
  8. from oas.ease.exceptions import *
  9. from oas.ease.response import *
  10. from oas.ease.utils import *
  11. from oas.ease.vault import *
  12. from oas.ease.uploader import *
  13. from oas.ease.job import *
  15. import os
  17. LOG_FILE="test.log"
  18. handler = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes = 1024*1024, backupCount = 5)
  19. fmt = '%(asctime)s - %(filename)s:%(lineno)s - %(name)s - %(message)s'  
  20. formatter = logging.Formatter(fmt)
  21. handler.setFormatter(formatter)
  22. log.addHandler(handler)
  23. log.setLevel(logging.DEBUG)
  26. class TestConf(object):
  28.     def __init__(self):
  29.         self.host = 'oas域名'
  30.         self.accesskey_id = "您的access key Id"
  31.         self.accesskey_secret = "您的access key secret"
  32.         self.vault_name = "normal"
  33.         self.vault_name_test = "test"
  35.         self.osshost = "您要转储的oss域名"
  36.         self.bucket = "您要转储的bucket"
  37.         self.object = "您要转储的Object"
  39. conf = TestConf()
  42. class TestDemo():
  43.     _MEGABYTE = 1024 * 1024
  46.     def __init__(self):
  47.         self.api = OASAPI(conf.host, conf.accesskey_id, conf.accesskey_secret)
  48.         self.api_proxy = APIProxy(self.api)
  49.         self.vault_name = conf.vault_name
  50.         self.vault_name_test = conf.vault_name_test
  52.         self.file_big = "random100M.bin"
  53.         self.file_big_size = 200*self._MEGABYTE
  55.         with open(self.file_big, 'wb+') as f:
  56.             self.write_random_data(f, self.file_big_size)
  58.         self.file_small = "random30M.bin"
  59.         self.file_small_size = 30*self._MEGABYTE
  61.         with open(self.file_small, 'wb+') as f:
  62.             self.write_random_data(f, self.file_small_size)
  65.     def write_random_data(self, file, size):
  66.         remaining = size
  67.         while remaining > 0:
  68.             n = min(remaining, 1024)
  69.             random_data = os.urandom(n)
  70.             file.write(random_data)
  71.             remaining = remaining - n
  73.             n = min(remaining, 1024 * random.randint(256, 1024))
  74.             file.write('\0' * n)
  75.             remaining = remaining - n
  78.     def test_single_archive_upload(self):
  79.         print '''
  80. ########################################################
  81.  Using the low level api method to upload a small archive 
  82. #########################################################'''
  83.         res = self.api_proxy.create_vault(self.vault_name)
  84.         vault_id = res['x-oas-vault-id']
  86.         etag = compute_etag_from_file(self.file_small)
  87.         tree_etag = compute_tree_etag_from_file(self.file_small)
  88.         with open(self.file_small, 'rb') as f:
  89.             content = f.read()
  90.             res = self.api_proxy.post_archive(vault_id, content, etag, tree_etag)
  91.             archive_id = res['x-oas-archive-id']
  92.         print "archive_id",archive_id
  94.     #test the multipart upload by the proxy api, and this is low level api
  95.     def test_multi_upload(self):
  96.         print '''\n\n\n
  97. ########################################################
  98.  Using the low level api to invoke the multipart api and implement the archive uplod
  99. #########################################################'''
  100.         res = self.api_proxy.create_vault(self.vault_name)
  101.         vault_id = res['x-oas-vault-id']
  103.         part_size = 1024 * 1024 * 64
  106.         etag_array= []
  107.         tree_etag_array= []
  108.         offset = 0
  110.         cur_part_num=0
  111.         cur_start_pos = 0
  112.         cur_end_pos = 0
  113.         print "1. comput the etag,tree-etag"
  114.         print "1.1 comput the  etag , tree_etag of part" 
  115.         while True:
  116.             tmpsize = part_size
  117.             if(cur_start_pos + tmpsize > self.file_big_size):
  118.                 tmpsize = self.file_big_size - cur_start_pos;
  119.             cur_end_pos += tmpsize -1
  121.             etag_array.append(compute_etag_from_file(self.file_big, cur_start_pos, tmpsize) )
  122.             tree_etag_array.append(compute_tree_etag_from_file(self.file_big, cur_start_pos, tmpsize))
  124.             cur_start_pos += tmpsize
  125.             cur_part_num += 1
  126.             if(cur_start_pos >= self.file_big_size-1):
  127.                 break;
  129.         print "1.2 comput the  total tree_etag of the archive" 
  130.         tree_etag_total = compute_combine_tree_etag_from_list( tree_etag_array) ;
  132.         print "2.1 init the upload task, and get the uploadId"
  133.         res = self.api_proxy.create_multipart_upload(vault_id, part_size)
  134.         upload_id = res['x-oas-multipart-upload-id']
  135.         print "upload_id",upload_id
  137.         f = open(self.file_big, 'rb')
  139.         cur_part_num = 0
  140.         cur_start_pos = 0
  141.         cur_end_pos = 0
  142.         while True:
  143.             tmpsize = part_size
  144.             if(cur_start_pos + tmpsize > self.file_big_size):
  145.                 tmpsize = self.file_big_size - cur_start_pos;
  147.             cur_end_pos += tmpsize
  148.             print "2.2 upload every part to oas server, and the etag of the part will be used. current part is:", cur_part_num+1
  149.             self.api_proxy.post_multipart_from_reader(vault_id, upload_id, f, tmpsize,('%d-%d' %(cur_start_pos, cur_end_pos-1)), etag_array[cur_part_num],tree_etag_array[cur_part_num])
  151.             cur_start_pos += tmpsize
  152.             cur_part_num += 1
  153.             if(cur_end_pos>= self.file_big_size -1):
  154.                 break;
  156.         print "2.3 complete the multipart task, and the total etag will be used"
  157.         res = self.api_proxy.complete_multipart_upload(
  158.                         vault_id, upload_id, self.file_big_size, tree_etag_total)
  159.         print "output the archive id"
  160.         archive_id = res['x-oas-archive-id']
  161.         print "archive_id",archive_id
  162.         return archive_id
  164.     #test the uploader to upload big file, and this is high level api
  165.     def test_uploader(self):
  166.         print '''\n\n\n
  167. ########################################################
  168.  Using the High level api to invoke the multipart api 
  169. #########################################################'''
  170.         vault = Vault.create_vault(self.api, self.vault_name)
  172.         print "initial a uploadId"
  173.         uploader = vault.initiate_uploader(self.file_big)
  174.         uploader_id = uploader.id
  175.         print "uploader_id",uploader_id
  177.         print "start the multipart" 
  178.         archive_id = uploader.start()
  180.         print "finish the upload, and output the archive_id"
  181.         print "archive_id",archive_id
  182.         return archive_id
  185.     #to inquire the archive list of vault
  186.     def test_vault_retrieve(self):
  187.         print '''\n\n\n
  188. ########################################################
  189.  Retrieve the vault info, and inquire the archive list of the vault 
  190. #########################################################'''
  191.         vault = Vault.create_vault(self.api, self.vault_name)
  192.         job = vault.retrieve_inventory()
  193.         job.download_to_file(job.id)
  195.     #to test download archive
  196.     def test_download_archive(self, archive_id):
  197.         print '''\n\n\n
  198. ########################################################
  199.  Submit the archive job and download the job to local 
  200. #########################################################'''
  201.         vault = Vault.create_vault(self.api, self.vault_name)
  202.         job = vault.retrieve_archive(archive_id)
  203.         job.update_status()
  204.         if not job.completed:
  205.             job.download_to_file(file_path = job.id, block = True)
  207.     #test delete archive
  208.     def test_delete_archive(self, archive_id):
  209.         print '''\n\n\n
  210. ########################################################
  211.  Delete the archive 
  212. #########################################################'''
  213.         vault = Vault.create_vault(self.api, self.vault_name)
  214.         vault.delete_archive(archive_id)
  216.     def test_pull_from_oss(self):
  217.         print '''\n\n\n
  218. ########################################################
  219.  submit a pull-from-oss job and OAS will finish the pull of OSS object to OAS 
  220. #########################################################'''
  221.         vault = Vault.create_vault(self.api, self.vault_name)
  222.         # create vault
  223.         job = vault.pull_from_oss(conf.osshost, conf.bucket, conf.object, "test desc")
  224.         job._check_status(block=True)
  226.         # delete archive
  227.         print "bucket:%s object:%s finish pull from oss,\n archiveId:%s" % (conf.bucket, conf.object, job.archive_id)
  230.     def test_push_to_oss(self):
  231.         print '''\n\n\n
  232. ########################################################
  233.  submit a push-to-oss job and OAS will finish the push of OAS archive to OSS 
  234. #########################################################'''
  235.         vault = Vault.create_vault(self.api, self.vault_name)
  236.         archive_id = vault.upload_archive(self.file_big)
  237.         job = vault.push_to_oss(archive_id, conf.osshost, conf.bucket, archive_id, "test desc")
  238.         job._check_status(block=True)
  239.         print archive_id + " finish push to oss"        
  241. if __name__ == '__main__':
  242.     t = TestDemo()
  244.     t.test_single_archive_upload()
  245.     t.test_multi_upload();
  247.     archive_id = t.test_uploader();
  249.     t.test_vault_retrieve();
  250.     t.test_download_archive(archive_id)
  251.     t.test_delete_archive(archive_id)
  253.     t.test_pull_from_oss()
  254.     t.test_push_to_oss()