Merge pull request #267 from l-iberty/master

功能特性更新

Author: l-iberty

qcloud_cos/cos_client.py

@@ -44,7 +44,7 @@ def __init__(self, Appid=None, Region=None, SecretId=None, SecretKey=None, Token
                  Access_id=None, Access_key=None, Secret_id=None, Secret_key=None, Endpoint=None, IP=None, Port=None,
                  Anonymous=None, UA=None, Proxies=None, Domain=None, ServiceDomain=None, KeepAlive=True, PoolConnections=10,
                  PoolMaxSize=10, AllowRedirects=False, SignHost=True, EndpointCi=None, EndpointPic=None, EnableOldDomain=True, EnableInternalDomain=True, SignParams=True,
-                 AutoSwitchDomainOnRetry=True):
+                 AutoSwitchDomainOnRetry=False):
         """初始化，保存用户的信息
 
         :param Appid(string): 用户APPID.
@@ -359,6 +359,7 @@ def send_request(self, method, url, bucket, timeout=30, cos_request=True, ci_req
             kwargs['verify'] = False
         if self._conf._allow_redirects is not None:
             kwargs['allow_redirects'] = self._conf._allow_redirects
+        exception_logbuf = list() # 记录每次重试的错误日志
         for j in range(self._retry + 1):
             try:
                 if j != 0:
@@ -397,13 +398,16 @@ def send_request(self, method, url, bucket, timeout=30, cos_request=True, ci_req
                     else:
                         break
             except Exception as e:  # 捕获requests抛出的如timeout等客户端错误,转化为客户端错误
-                logger.exception('url:%s, retry_time:%d exception:%s' % (url, j, str(e)))
+                # 记录每次请求的exception
+                exception_log = 'url:%s, retry_time:%d exception:%s' % (url, j, str(e))
+                exception_logbuf.append(exception_log)
                 if j < self._retry and (isinstance(e, ConnectionError) or isinstance(e, Timeout)):  # 只重试网络错误
                     if client_can_retry(file_position, **kwargs):
                         if not domain_switched and self._conf._auto_switch_domain_on_retry and self._conf._ip is None:
                             url = switch_hostname_for_url(url)
                             domain_switched = True
                         continue
+                logger.exception(exception_logbuf) # 最终重试失败, 输出前几次重试失败的exception
                 raise CosClientError(str(e))
 
         if not cos_request:
@@ -419,12 +423,16 @@ def send_request(self, method, url, bucket, timeout=30, cos_request=True, ci_req
                 if 'x-cos-trace-id' in res.headers:
                     info['traceid'] = res.headers['x-cos-trace-id']
                 logger.warn(info)
+                if len(exception_logbuf) > 0:
+                    logger.exception(exception_logbuf) # 最终重试失败, 输出前几次重试失败的exception
                 raise CosServiceError(method, info, res.status_code)
             else:
                 msg = res.text
                 if msg == u'':  # 服务器没有返回Error Body时 给出头部的信息
                     msg = res.headers
                 logger.error(msg)
+                if len(exception_logbuf) > 0:
+                    logger.exception(exception_logbuf) # 最终重试失败, 输出前几次重试失败的exception
                 raise CosServiceError(method, msg, res.status_code)
 
         return None
@@ -1613,7 +1621,7 @@ def head_bucket(self, Bucket, **kwargs):
 
         :param Bucket(string): 存储桶名称.
         :param kwargs(dict): 设置请求headers.
-        :return: None.
+        :return: HEAD Bucket响应头域.
 
         .. code-block:: python
 
@@ -1635,7 +1643,7 @@ def head_bucket(self, Bucket, **kwargs):
             bucket=Bucket,
             auth=CosS3Auth(self._conf),
             headers=headers)
-        return None
+        return rt.headers
 
     def put_bucket_acl(self, Bucket, AccessControlPolicy={}, **kwargs):
         """设置bucket ACL
@@ -2926,7 +2934,7 @@ def get_bucket_inventory(self, Bucket, Id, **kwargs):
         return data
 
     def delete_bucket_inventory(self, Bucket, Id, **kwargs):
-        """删除bucket 回源配置
+        """删除bucket清单规则
 
         :param Bucket(string): 存储桶名称.
         :param Id(string): 清单规则名称.
@@ -2957,6 +2965,52 @@ def delete_bucket_inventory(self, Bucket, Id, **kwargs):
             params=params)
         return None
 
+    def list_bucket_inventory_configurations(self, Bucket, ContinuationToken=None, **kwargs):
+        """列举存储桶清单规则
+
+        :param Bucket(string): 存储桶名称
+        :param ContinuationToken(string): 分页参数, 用以获取下一页信息
+        :param kwargs(dict): 设置请求headers.
+        :return(dict): 存储桶清单规则列表
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 分页列举bucket清单规则
+            continuation_token = ''
+            while True:
+                resp = client.list_bucket_inventory_configurations(
+                    Bucket=bucket,
+                    ContinuationToken=continuation_token,
+                )
+                if 'InventoryConfiguration' in resp:
+                    for conf in resp['InventoryConfiguration']:
+                        print(conf)
+                if resp['IsTruncated'] == 'true':
+                    continuation_token = resp['NextContinuationToken']
+                else:
+                    break
+        """
+        headers = mapped(kwargs)
+        params = {'inventory': ''}
+        if ContinuationToken is not None:
+            params['continuation-token'] = ContinuationToken
+        url = self._conf.uri(bucket=Bucket)
+        logger.info("list bucket inventory configurations, url={url}, headers={headers}".format(
+            url=url,
+            headers=headers,
+        ))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            bucket=Bucket,
+            auth=CosS3Auth(self._conf, params=params),
+            headers=headers,
+            params=params)
+        data = xml_to_dict(rt.content)
+        return data
+
     def put_object_tagging(self, Bucket, Key, Tagging={}, **kwargs):
         """设置object的标签
 
@@ -3560,7 +3614,7 @@ def _check_all_upload_parts(self, bucket, key, uploadid, local_path, parts_num,
             already_exist_parts[part_num] = part['ETag']
         return True
 
-    def download_file(self, Bucket, Key, DestFilePath, PartSize=20, MAXThread=5, EnableCRC=False, progress_callback=None, **Kwargs):
+    def download_file(self, Bucket, Key, DestFilePath, PartSize=20, MAXThread=5, EnableCRC=False, progress_callback=None, DumpRecordDir=None, **Kwargs):
         """小于等于20MB的文件简单下载，大于20MB的文件使用续传下载
 
         :param Bucket(string): 存储桶名称.
@@ -3596,7 +3650,7 @@ def download_file(self, Bucket, Key, DestFilePath, PartSize=20, MAXThread=5, Ena
             callback = ProgressCallback(file_size, progress_callback)
 
         downloader = ResumableDownLoader(self, Bucket, Key, DestFilePath, object_info, PartSize, MAXThread, EnableCRC,
-                                         callback, **Kwargs)
+                                         callback, DumpRecordDir, **Kwargs)
         downloader.start()
 
     def upload_file(self, Bucket, Key, LocalFilePath, PartSize=1, MAXThread=5, EnableMD5=False, progress_callback=None,

qcloud_cos/resumable_downloader.py

@@ -17,7 +17,7 @@
 
 class ResumableDownLoader(object):
     def __init__(self, cos_client, bucket, key, dest_filename, object_info, part_size=20, max_thread=5,
-                 enable_crc=False, progress_callback=None, **kwargs):
+                 enable_crc=False, progress_callback=None, dump_record_dir = None, **kwargs):
         self.__cos_client = cos_client
         self.__bucket = bucket
         self.__key = key
@@ -34,7 +34,10 @@ def __init__(self, cos_client, bucket, key, dest_filename, object_info, part_siz
         self.__finished_parts = []
         self.__lock = threading.Lock()
         self.__record = None  # 记录当前的上下文
-        self.__dump_record_dir = os.path.join(os.path.expanduser('~'), '.cos_download_tmp_file')
+        if not dump_record_dir:
+            self.__dump_record_dir = os.path.join(os.path.expanduser('~'), '.cos_download_tmp_file')
+        else:
+            self.__dump_record_dir = dump_record_dir
 
         record_filename = self.__get_record_filename(bucket, key, self.__dest_file_path)
         self.__record_filepath = os.path.join(self.__dump_record_dir, record_filename)

ut/test.py

@@ -1506,6 +1506,76 @@ def test_put_get_delete_bucket_inventory():
         Id='test'
     )
 
+def test_list_bucket_inventory_configrations():
+    """测试列举bucket清单"""
+    inventory_config = {
+        'Destination': {
+            'COSBucketDestination': {
+                'AccountId': '2832742109',
+                'Bucket': 'qcs::cos:' + REGION + '::' + test_bucket,
+                'Format': 'CSV',
+                'Prefix': 'list1',
+                'Encryption': {
+                    'SSECOS': {}
+                }
+            }
+        },
+        'IsEnabled': 'True',
+        'Filter': {
+            'Prefix': 'filterPrefix'
+        },
+        'IncludedObjectVersions': 'All',
+        'OptionalFields': {
+            'Field': [
+                'Size',
+                'LastModifiedDate',
+                'ETag',
+                'StorageClass',
+                'IsMultipartUploaded',
+                'ReplicationStatus'
+            ]
+        },
+        'Schedule': {
+            'Frequency': 'Daily'
+        }
+    }
+    # 构建150条清单配置规则(清单分页大小为100条规则)
+    n = 150
+    for i in range(n):
+        id = 'ID-{}'.format(i)
+        response = client.put_bucket_inventory(
+            Bucket=test_bucket,
+            Id=id,
+            InventoryConfiguration=inventory_config,
+        )
+    
+    # 列举清单
+    i = 0
+    continuation_token = ''
+    while True:
+        resp = client.list_bucket_inventory_configurations(
+            Bucket=test_bucket,
+            ContinuationToken=continuation_token,
+        )
+        if 'InventoryConfiguration' in resp:
+            for conf in resp['InventoryConfiguration']:
+                id = 'ID-{}'.format(i)
+                assert id == conf['Id']
+                i += 1
+        if resp['IsTruncated'] == 'true':
+            continuation_token = resp['NextContinuationToken']
+        else:
+            break
+    
+    assert i == n
+
+    # 删除清单
+    for i in range(n):
+        id = 'ID-{}'.format(i)
+        response = client.delete_bucket_inventory(
+            Bucket=test_bucket,
+            Id=id,
+        )
 
 def test_put_get_delete_bucket_tagging():
     """测试设置获取删除bucket标签"""