head_bucket添加返回值; list_bucket_inventory_configurations接口

Author: libertyzhu

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.
@@ -1635,7 +1635,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 +2926,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 +2957,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']:
+                        id = 'ID-{}'.format(i)
+                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的标签
 

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标签"""