get_bucket_meta接口

Author: libertyzhu

qcloud_cos/cos_client.py

@@ -3468,7 +3468,7 @@ def put_bucket_intelligenttiering(self, Bucket, IntelligentTieringConfiguration=
         """设置存储桶智能分层配置
 
         :param Bucket(string): 存储桶名称.
-        :param IntelligentTieringConfiguration(dict): 只能分层配置
+        :param IntelligentTieringConfiguration(dict): 智能分层配置
         :param kwargs(dict): 设置请求headers.
         :return: None.
 
@@ -3510,7 +3510,6 @@ def put_bucket_intelligenttiering(self, Bucket, IntelligentTieringConfiguration=
     def get_bucket_intelligenttiering(self, Bucket, **kwargs):
         """获取存储桶智能分层配置
         :param Bucket(string): 存储桶名称.
-        :param IntelligentTieringConfiguration(dict): 智能分层配置
         :param kwargs(dict): 设置请求headers.
         :return(dict): 智能分层配置.
 
@@ -3535,6 +3534,225 @@ def get_bucket_intelligenttiering(self, Bucket, **kwargs):
             params=params)
         data = xml_to_dict(rt.content)
         return data
+    
+    def put_bucket_object_lock(self, Bucket, ObjectLockConfiguration={}, **kwargs):
+        """设置存储桶对象锁定配置
+
+        :param Bucket(string): 存储桶名称.
+        :param ObjectLockConfiguration(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)
+
+            object_lock_conf = {
+                'ObjectLockEnabled': 'Enabled',
+            }
+            client.put_bucket_object_lock(Bucket="bucket", ObjectLockConfiguration=objeck_lock_conf)
+        """
+
+        xml_config = format_xml(data=ObjectLockConfiguration, root='ObjectLockConfiguration')
+        headers = mapped(kwargs)
+        headers['Content-MD5'] = get_md5(xml_config)
+        headers['Content-Type'] = 'application/xml'
+        params = {'object-lock': ''}
+        url = self._conf.uri(bucket=Bucket)
+        logger.info("put bucket object-lock, 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 rt.headers
+    
+    def get_bucket_object_lock(self, Bucket, **kwargs):
+        """获取存储桶对象锁定配置
+
+        :param Bucket(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)
+            client.get_bucket_object_lock(Bucket="bucket")
+        """
+        headers = mapped(kwargs)
+        params = {'object-lock': ''}
+        url = self._conf.uri(bucket=Bucket)
+        logger.info("get bucket object-lock, 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 get_bucket_meta(self, Bucket, **kwargs):
+        """获取存储桶各项配置
+
+        :param Bucket(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)
+            client.get_bucket_meta(Bucket="bucket")
+        """
+        data = {
+            'BucketUrl': None,
+            'OFS': False,
+            'MAZ': False,
+            'Encryption': None,
+            'ACL': None,
+            'Website': None,
+            'Logging': None,
+            'CORS': None,
+            'Versioning': None,
+            'IntelligentTiering': None,
+            'Lifecycle': None,
+            'Tagging': None,
+            'ObjectLock': None,
+            'Replication': None,
+        }
+        pool = SimpleThreadPool(num_threads=10)
+
+        # HeadBucket
+        def _head_bucket_wrapper(Bucket, **kwargs):
+            resp = self.head_bucket(Bucket, **kwargs)
+            # x-cos-bucket-arch: 'OFS'
+            # x-cos-bucket-az-type: 'MAZ'
+            # x-cos-bucket-region: 'ap-beijing'
+            if 'x-cos-bucket-arch' in resp and resp['x-cos-bucket-arch'] == 'OFS':
+                data.update({'OFS': True})
+            else:
+                data.update({'OFS': False})
+            if 'x-cos-bucket-az-type' in resp and resp['x-cos-bucket-az-type'] == 'MAZ':
+                data.update({'MAZ': True})
+            else:
+                data.update({'MAZ': False})
+            data.update({"Location": resp['x-cos-bucket-region']})
+            url = self._conf.uri(bucket=Bucket)
+            data.update({'BucketUrl': url})
+        pool.add_task(_head_bucket_wrapper, Bucket, **kwargs)
+
+        # Website
+        def _get_bucket_website_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_website(Bucket, **kwargs)
+                data.update({'Website': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get website conf:{}".format(e))
+        pool.add_task(_get_bucket_website_wrapper, Bucket, **kwargs)
+
+        # ObjectLock
+        def _get_bucket_object_lock_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_object_lock(Bucket, **kwargs)
+                data.update({'ObjectLock': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get object_lock conf:{}".format(e))
+        pool.add_task(_get_bucket_object_lock_wrapper, Bucket, **kwargs)
+
+        # ACL
+        def _get_bucket_acl_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_acl(Bucket, **kwargs)
+                data.update({'ACL': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get acl conf:{}".format(e))
+        pool.add_task(_get_bucket_acl_wrapper, Bucket, **kwargs)
+
+        # Logging
+        def _get_bucket_logging_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_logging(Bucket, **kwargs)
+                data.update({'Logging': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get logging conf:{}".format(e))
+        pool.add_task(_get_bucket_logging_wrapper, Bucket, **kwargs)
+
+        # Lifecycle
+        def _get_bucket_lifecycle_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_lifecycle(Bucket, **kwargs)
+                data.update({'Lifecycle': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get lifecycle conf:{}".format(e))
+        pool.add_task(_get_bucket_lifecycle_wrapper, Bucket, **kwargs)
+
+        # Replication
+        def _get_bucket_replication_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_replication(Bucket, **kwargs)
+                data.update({'Replication': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get replication conf:{}".format(e))
+        pool.add_task(_get_bucket_replication_wrapper, Bucket, **kwargs)
+
+        # Replication
+        def _get_bucket_encryption_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_encryption(Bucket, **kwargs)
+                data.update({'Encryption': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get encryption conf:{}".format(e))
+        pool.add_task(_get_bucket_encryption_wrapper, Bucket, **kwargs)
+
+        # CORS
+        def _get_bucket_cors_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_cors(Bucket, **kwargs)
+                data.update({'CORS': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get cors conf:{}".format(e))
+        pool.add_task(_get_bucket_cors_wrapper, Bucket, **kwargs)
+
+        # Versioning
+        def _get_bucket_versioning_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_versioning(Bucket, **kwargs)
+                data.update({'Versioning': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get versioning conf:{}".format(e))
+        pool.add_task(_get_bucket_versioning_wrapper, Bucket, **kwargs)
+
+        # IntelligentTiering
+        def _get_bucket_intelligenttiering_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_intelligenttiering(Bucket, **kwargs)
+                data.update({'IntelligentTiering': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get intelligenttiering conf:{}".format(e))
+        pool.add_task(_get_bucket_intelligenttiering_wrapper, Bucket, **kwargs)
+
+        # Tagging
+        def _get_bucket_tagging_wrapper(Bucket, **kwargs):
+            try:
+                resp = self.get_bucket_tagging(Bucket, **kwargs)
+                data.update({'Tagging': resp})
+            except CosServiceError as e:
+                logger.debug("get_bucket_meta failed to get tagging conf:{}".format(e))
+        pool.add_task(_get_bucket_tagging_wrapper, Bucket, **kwargs)
+
+        pool.wait_completion()
+        return data
 
     # service interface begin
     def list_buckets(self, TagKey=None, TagValue=None, Region=None, CreateTime=None, Range=None, Marker="", MaxKeys=2000, **kwargs):

ut/test.py

@@ -30,6 +30,8 @@
 USE_CREDENTIAL_INST = os.environ["USE_CREDENTIAL_INST"]
 test_bucket = 'cos-python-v5-test-' + str(sys.version_info[0]) + '-' + str(
     sys.version_info[1]) + '-' + REGION + '-' + APPID
+test_worm_bucket = 'cos-python-v5-test-worm' + str(sys.version_info[0]) + '-' + str(
+    sys.version_info[1]) + '-' + REGION + '-' + APPID
 copy_test_bucket = 'copy-' + test_bucket
 test_object = "test.txt"
 special_file_name = "中文" + \
@@ -761,6 +763,54 @@ def test_get_bucket_location():
     )
     assert response['LocationConstraint'] == REGION
 
+def test_put_get_bucket_object_lock():
+    """bucket object_lock测试"""
+    
+    # 创建worm测试桶
+    try:
+        response = client.create_bucket(Bucket=test_worm_bucket)
+    except CosServiceError as e:
+        error_code = e.get_error_code()
+        if error_code == 'BucketAlreadyOwnedByYou' or error_code == 'BucketAlreadyExists':
+            pass
+        else:
+            raise e
+    
+    object_lock_conf = {
+        'ObjectLockEnabled': 'Enabled',
+    }
+    response = client.put_bucket_object_lock(Bucket=test_worm_bucket, ObjectLockConfiguration=object_lock_conf)
+
+    time.sleep(3)
+    response = client.get_bucket_object_lock(Bucket=test_worm_bucket)
+    assert response
+    assert response['ObjectLockEnabled'] == 'Enabled'
+
+    # test get_bucket_meta() by the way.
+    meta = client.get_bucket_meta(Bucket=test_worm_bucket)
+    assert meta
+
+    # 删除worm测试桶
+    client.delete_bucket(Bucket=test_worm_bucket)
+
+def test_get_bucket_meta():
+    """测试get_buckt_meta()接口"""
+    response = client.get_bucket_meta(Bucket=test_bucket)
+    assert response
+    assert 'BucketUrl' in response
+    assert 'OFS' in response
+    assert 'MAZ' in response
+    assert 'Encryption' in response
+    assert 'ACL' in response
+    assert 'Website' in response
+    assert 'Logging' in response
+    assert 'CORS' in response
+    assert 'Versioning' in response
+    assert 'IntelligentTiering' in response
+    assert 'Lifecycle' in response
+    assert 'Tagging' in response
+    assert 'ObjectLock' in response
+    assert 'Replication' in response
 
 def test_get_service():
     """列出账号下所有的bucket信息"""