Add bucket storage capacity api

examples/bucket_user_qos.py

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+
+import os
+import oss2
+
+# 以下代码展示了set_bucket_storage_capacity的的用法。
+
+# 首先初始化AccessKeyId、AccessKeySecret、Endpoint等信息。
+# 通过环境变量获取，或者把诸如“<你的AccessKeyId>”替换成真实的AccessKeyId等。
+access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '<你的AccessKeyId>')
+access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '<你的AccessKeySecret>')
+bucket_name = os.getenv('OSS_TEST_BUCKET', '<你的Bucket>')
+endpoint = os.getenv('OSS_TEST_ENDPOINT', '<你的访问域名>')
+
+# 确认上面的参数都填写正确了
+for param in (access_key_id, access_key_secret, bucket_name, endpoint):
+    assert '<' not in param, '请设置参数：' + param
+
+# 创建Bucket对象，所有Object相关的接口都可以通过Bucket对象来进行
+bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
+
+# 设置bucket容量为100GB
+user_qos = oss2.models.BucketUserQos(100)
+resp = bucket.set_bucket_storage_capacity(user_qos)
+
+# 获取容量信息
+result = bucket.get_bucket_storage_capacity()
+print("bucket_qos_capacity:" + result.storage_capacity)

oss2/api.py

@@ -384,6 +384,7 @@ class Bucket(_Base):
     POLICY = 'policy'
     REQUESTPAYMENT  = 'requestPayment'
     QOS_INFO = 'qosInfo'
+    USER_QOS = 'qos'
 
     def __init__(self, auth, endpoint, bucket_name,
                  is_cname=False,
@@ -2173,6 +2174,23 @@ def delete_bucket_qos_info(self):
 
         return RequestResult(resp)
 
+    def set_bucket_storage_capacity(self, user_qos):
+        """设置Bucket的容量，单位GB
+
+        :param user_qos :class:`BucketUserQos <oss2.models.BucketUserQos>`
+        """
+        data = xml_utils.to_put_bucket_user_qos(user_qos)
+        resp = self.__do_bucket('PUT', data=data, params={Bucket.USER_QOS: ''})
+        return RequestResult(resp)
+
+    def get_bucket_storage_capacity(self):
+        """获取bucket的容量信息。
+
+        :return: :class:`GetBucketUserQosResult <oss2.models.GetBucketUserQosResult>`
+        """
+        resp = self._Bucket__do_bucket('GET', params={Bucket.USER_QOS: ''})
+        return self._parse_result(resp, xml_utils.parse_get_bucket_user_qos, GetBucketUserQosResult)
+
     def _get_bucket_config(self, config):
         """获得Bucket某项配置，具体哪种配置由 `config` 指定。该接口直接返回 `RequestResult` 对象。
         通过read()接口可以获得XML字符串。不建议使用。

oss2/models.py

@@ -1455,3 +1455,15 @@ class GetBucketQosInfoResult(RequestResult, BucketQosInfo):
     def __init__(self, resp):
         RequestResult.__init__(self, resp)
         BucketQosInfo.__init__(self)
+
+class BucketUserQos(object):
+    """用户服务质量。
+    :param int storage_capacity: 容量大小，单位GB
+    """
+    def __init__(self, storage_capacity=None):
+        self.storage_capacity = storage_capacity
+
+class GetBucketUserQosResult(RequestResult, BucketUserQos):
+    def __init__(self, resp):
+        RequestResult.__init__(self, resp)
+        BucketUserQos.__init__(self)

oss2/xml_utils.py

@@ -1229,3 +1229,17 @@ def parse_get_qos_info(result, body):
     result.extranet_qps = _find_int(root, 'ExtranetQps')
 
     return result
+
+def parse_get_bucket_user_qos(result, body):
+    root = ElementTree.fromstring(body)
+
+    result.storage_capacity = _find_int(root, 'StorageCapacity')
+
+    return result
+
+def to_put_bucket_user_qos(user_qos):
+    root = ElementTree.Element('BucketUserQos')
+
+    _add_text_child(root, 'StorageCapacity', str(user_qos.storage_capacity))
+
+    return _node_to_string(root)

tests/test_bucket.py

@@ -849,6 +849,33 @@ def test_bucket_policy(self):
         self.assertEqual(int(result.status)//100, 2)
         bucket.delete_bucket()
 
+    def test_bucket_storage_capacity(self):
+        auth = oss2.Auth(OSS_ID, OSS_SECRET)
+        bucket = oss2.Bucket(auth, OSS_ENDPOINT, random_string(63).lower())
+        bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)
+
+        # test default storage capacity
+        result = bucket.get_bucket_storage_capacity()
+        self.assertEqual(result.storage_capacity, -1)
+
+        #test set default capacity value -1
+        user_qos = oss2.models.BucketUserQos(-1)
+        bucket.set_bucket_storage_capacity(user_qos)
+        result = bucket.get_bucket_storage_capacity()
+        self.assertEqual(result.storage_capacity, -1)
+
+        #set neagetive value other than -1
+        user_qos = oss2.models.BucketUserQos(-2)
+        self.assertRaises(oss2.exceptions.InvalidArgument, bucket.set_bucket_storage_capacity, user_qos) 
+
+        #set positive value
+        user_qos = oss2.models.BucketUserQos(100)
+        resp = bucket.set_bucket_storage_capacity(user_qos)
+        result = bucket.get_bucket_storage_capacity()
+        self.assertEqual(result.storage_capacity, 100)
+
+        bucket.delete_bucket()
+
     def test_malformed_xml(self):
         xml_input = '''<This is a bad xml></bad as I am>'''
         self.assertRaises(oss2.exceptions.MalformedXml, self.bucket.put_bucket_lifecycle, xml_input)