Feature intelligenttiering

demo/demo.py

@@ -17,8 +17,8 @@
 
 # 设置用户属性, 包括secret_id, secret_key, region
 # appid已在配置中移除,请在参数Bucket中带上appid。Bucket由bucketname-appid组成
-secret_id = 'AKID15IsskiBQACGbAo6WhgcQbVls7HmuG00'     # 替换为用户的secret_id
-secret_key = 'csivKvxxrMvSvQpMWHuIz12pThQQlWRW'     # 替换为用户的secret_key
+secret_id = 'secret_id'     # 替换为用户的secret_id
+secret_key = 'secret_key'     # 替换为用户的secret_key
 region = 'ap-beijing'    # 替换为用户的region
 token = None               # 使用临时密钥需要传入Token，默认为空,可不填
 config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象

qcloud_cos/cos_client.py

@@ -2861,6 +2861,7 @@ def delete_bucket_referer(self, Bucket, **kwargs):
                 Bucket='bucket'
             )
         """
+
         xml_config = ''
         headers = mapped(kwargs)
         headers['Content-MD5'] = get_md5(xml_config)
@@ -2877,6 +2878,78 @@ def delete_bucket_referer(self, Bucket, **kwargs):
             params=params)
         return None
 
+    def put_bucket_intelligenttiering(self, Bucket, IntelligentTieringConfiguration=None, **kwargs):
+        """设置存储桶智能分层配置
+
+        :param Bucket(string): 存储桶名称.
+        :param IntelligentTieringConfiguration(dict): 只能分层配置
+        :param kwargs(dict): 设置请求headers.
+        :return: None.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+
+            intelligent_tiering_conf = {
+                'Status': 'Enable',
+                'Transition': {
+                    'Days': '30|60|90',
+                    'RequestFrequent': '1'
+                }
+            }
+            client.put_bucket_intelligenttiering(Bucket="bucket", IntelligentTieringConfiguration=intelligent_tiering_conf)
+        """
+
+        if IntelligentTieringConfiguration is None:
+            IntelligentTieringConfiguration = {}
+        xml_config = format_xml(data=IntelligentTieringConfiguration, root='IntelligentTieringConfiguration')
+        headers = mapped(kwargs)
+        headers['Content-Type'] = 'application/xml'
+        params = {'intelligenttiering': ''}
+        url = self._conf.uri(bucket=Bucket)
+        logger.info("put bucket intelligenttiering, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='PUT',
+            url=url,
+            bucket=Bucket,
+            data=xml_config,
+            auth=CosS3Auth(self._conf, params=params),
+            headers=headers,
+            params=params)
+        return None
+
+    def get_bucket_intelligenttiering(self, Bucket, **kwargs):
+        """获取存储桶智能分层配置
+        :param Bucket(string): 存储桶名称.
+        :param IntelligentTieringConfiguration(dict): 只能分层配置
+        :param kwargs(dict): 设置请求headers.
+        :return(dict): 智能分层配置.
+
+        .. code-block:: python
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            client.get_bucket_intelligenttiering(Bucket='bucket')
+        """
+
+        headers = mapped(kwargs)
+        params = {'intelligenttiering': ''}
+        url = self._conf.uri(bucket=Bucket)
+        logger.info("get bucket intelligenttiering, 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
+
     # service interface begin
     def list_buckets(self, **kwargs):
         """列出所有bucket

qcloud_cos/demo.py

@@ -17,8 +17,8 @@
 
 # 设置用户属性, 包括secret_id, secret_key, region
 # appid已在配置中移除,请在参数Bucket中带上appid。Bucket由bucketname-appid组成
-secret_id = 'AKID15IsskiBQACGbAo6WhgcQbVls7HmuG00'     # 替换为用户的secret_id
-secret_key = 'csivKvxxrMvSvQpMWHuIz12pThQQlWRW'     # 替换为用户的secret_key
+secret_id = "secret_id"  # 替换为用户的secret_id
+secret_key = "secret_key"  # 替换为用户的secret_key
 region = 'ap-beijing-1'    # 替换为用户的region
 token = None               # 使用临时密钥需要传入Token，默认为空,可不填
 config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象

ut/test.py

@@ -1217,6 +1217,7 @@ def _test_get_object_sensitive_content_recognition():
     print(response)
     assert response
 
+
 def test_download_file():
     """测试断点续传下载接口"""
     #测试普通下载
@@ -1265,6 +1266,26 @@ def test_download_file():
     if os.path.exists(file_name):
         os.remove(file_name)
 
+
+def test_put_get_bucket_intelligenttiering():
+    """测试设置获取智能分层"""
+    intelligent_tiering_conf = {
+                'Status': 'Enable',
+                'Transition': {
+                    'Days': '30',
+                    'RequestFrequent': '1'
+                }
+            }
+    response = client.put_bucket_intelligenttiering(
+        Bucket=test_bucket,
+        IntelligentTieringConfiguration=intelligent_tiering_conf
+    )
+    time.sleep(2)
+    response = client.get_bucket_intelligenttiering(
+        Bucket=test_bucket,
+    )
+
+
 def test_bucket_encryption():
     """测试存储桶默认加密配置"""
     # 测试设置存储桶的默认加密配置
@@ -1287,6 +1308,7 @@ def test_bucket_encryption():
     # 删除存储桶默认加密配置
     client.delete_bucket_encryption(test_bucket)
 
+
 def test_aes_client():
     """测试aes加密客户端的上传下载操作"""
     content = '123456' * 1024 + '1'
@@ -1333,7 +1355,7 @@ def test_aes_client():
     assert local_file_md5 and content_md5 and local_file_md5 == content_md5
     if os.path.exists('test_multi_upload_local'):
         os.remove('test_multi_upload_local')
-    
+
     client_for_rsa.delete_object(test_bucket, 'test_multi_upload')
 
 def test_rsa_client():
@@ -1362,7 +1384,7 @@ def test_rsa_client():
     assert local_file_md5 and content_md5 and local_file_md5 == content_md5
     if os.path.exists('test_for_rsa_local'):
         os.remove('test_for_rsa_local')
-    
+
     client_for_rsa.delete_object(test_bucket, 'test_for_rsa')
 
     content = '1' * 1024 * 1024
@@ -1382,7 +1404,7 @@ def test_rsa_client():
     assert local_file_md5 and content_md5 and local_file_md5 == content_md5
     if os.path.exists('test_multi_upload_local'):
         os.remove('test_multi_upload_local')
-    
+
     client_for_rsa.delete_object(test_bucket, 'test_multi_upload')
 
 
@@ -1511,6 +1533,7 @@ def test_live_channel():
     response = client.delete_live_channel(Bucket=test_bucket, ChannelName=channel_name)
     assert (response)
 
+
 if __name__ == "__main__":
     setUp()
     """
@@ -1539,6 +1562,7 @@ def test_live_channel():
     _test_get_object_sensitive_content_recognition()
     test_live_channel()
     test_download_file()
+    test_put_get_bucket_intelligenttiering()
     test_aes_client()
     test_rsa_client()
     """