S3

README.rst

@@ -11,8 +11,7 @@ __________
 
 使用pip安装 ::
 
-    pip install -U qcloud_cos_v5  暂不可用
-
+    pip install -U cos-python-sdk-v5
 
 手动安装::
 

qcloud_cos/cos_client.py

@@ -82,6 +82,10 @@ def xml_to_dict(data):
     """V5使用xml格式，将response中的xml转换为dict"""
     root = xml.etree.ElementTree.fromstring(data)
     xmldict = Xml2Dict(root)
+    xmlstr = str(xmldict)
+    xmlstr = xmlstr.replace("{http://www.qcloud.com/document/product/436/7751}", "")
+    xmlstr = xmlstr.replace("{http://www.w3.org/2001/XMLSchema-instance}", "")
+    xmldict = eval(xmlstr)
     return xmldict
 
 
@@ -147,13 +151,11 @@ def __init__(self, conf, retry=1, session=None):
         else:
             self._session = session
 
-    def get_auth(self, Method, Bucket, Key=None, **kwargs):
+    def get_auth(self, Method, Bucket, Key=None, Expired=10000, headers={}, params={}):
         """获取签名"""
-        headers = mapped(kwargs)
-        # TODO(tiedu)  检查header的参数合法性
         url = self._conf.uri(bucket=Bucket, path=Key)
-        r = Request(Method, url)
-        auth = CosS3Auth(self._conf._access_id, self._conf._access_key)
+        r = Request(Method, url, headers=headers, params=params)
+        auth = CosS3Auth(self._conf._access_id, self._conf._access_key, Expired)
         return auth(r).headers['Authorization']
 
     def send_request(self, method, url, timeout=30, **kwargs):
@@ -239,10 +241,10 @@ def get_object(self, Bucket, Key, **kwargs):
             response[k] = rt.headers[k]
         return response
 
-    def get_presigned_download_url(self, Bucket, Key):
+    def get_presigned_download_url(self, Bucket, Key, Expired=10000):
         """生成预签名的下载url"""
         url = self._conf.uri(bucket=Bucket, path=Key)
-        sign = self.get_auth(Method='GET', Bucket=Bucket, Key=Key)
+        sign = self.get_auth(Method='GET', Bucket=Bucket, Key=Key, Expired=10000)
         url = url + '?sign=' + urllib.quote(sign)
         return url
 
@@ -353,11 +355,8 @@ def complete_multipart_upload(self, Bucket, Key, UploadId, MultipartUpload={}, *
                 data=dict_to_xml(MultipartUpload),
                 timeout=1200,  # 分片上传大文件的时间比较长，设置为20min
                 headers=headers)
-        response = dict()
         data = xml_to_dict(rt.text)
-        for key in data.keys():
-            response[key[key.find('}')+1:]] = data[key]
-        return response
+        return data
 
     def abort_multipart_upload(self, Bucket, Key, UploadId, **kwargs):
         """放弃一个已经存在的分片上传任务，删除所有已经存在的分片"""
@@ -385,18 +384,45 @@ def list_parts(self, Bucket, Key, UploadId, **kwargs):
                 url=url,
                 auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
                 headers=headers)
+        data = xml_to_dict(rt.text)
+        if 'Part' in data.keys() and isinstance(data['Part'], dict):  # 只有一个part，将dict转为list，保持一致
+            lst = []
+            lst.append(data['Part'])
+            data['Part'] = lst
+        return data
+
+    def put_object_acl(self, Bucket, Key, **kwargs):
+        """设置object ACL"""
+        headers = mapped(kwargs)
+        url = self._conf.uri(bucket=Bucket, path=Key+"?acl")
+        logger.info("put object acl, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='PUT',
+            url=url,
+            auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
+            headers=headers)
+        return None
 
+    def get_object_acl(self, Bucket, Key, **kwargs):
+        """获取object ACL"""
+        headers = mapped(kwargs)
+        url = self._conf.uri(bucket=Bucket, path=Key+"?acl")
+        logger.info("get object acl, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
+            headers=headers)
         data = xml_to_dict(rt.text)
-        if 'Part' in data.keys():
-            if isinstance(data['Part'], list):
-                return data
-            else:  # 只有一个part，将dict转为list，保持一致
-                lst = []
-                lst.append(data['Part'])
-                data['Part'] = lst
-                return data
-        else:
-            return data
+        if data['AccessControlList'] is not None and isinstance(data['AccessControlList']['Grant'], dict):
+            lst = []
+            lst.append(data['AccessControlList']['Grant'])
+            data['AccessControlList']['Grant'] = lst
+        return data
 
     # s3 bucket interface begin
     def create_bucket(self, Bucket, **kwargs):
@@ -447,16 +473,58 @@ def list_objects(self, Bucket, Delimiter="", Marker="", MaxKeys=1000, Prefix="",
                 auth=CosS3Auth(self._conf._access_id, self._conf._access_key))
 
         data = xml_to_dict(rt.text)
-        if 'Contents' in data.keys():
-            if isinstance(data['Contents'], list):
-                return data
-            else:  # 只有一个Contents，将dict转为list，保持一致
+        if 'Contents' in data.keys() and isinstance(data['Contents'], dict):  # 只有一个Contents，将dict转为list，保持一致
                 lst = []
                 lst.append(data['Contents'])
                 data['Contents'] = lst
-                return data
-        else:
-            return data
+        return data
+
+    def head_bucket(self, Bucket, **kwargs):
+        """获取bucket信息"""
+        headers = mapped(kwargs)
+        url = self._conf.uri(bucket=Bucket)
+        logger.info("head bucket, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='HEAD',
+            url=url,
+            auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
+            headers=headers)
+        return None
+
+    def put_bucket_acl(self, Bucket, **kwargs):
+        """设置bucket ACL"""
+        headers = mapped(kwargs)
+        url = self._conf.uri(bucket=Bucket, path="?acl")
+        logger.info("put bucket acl, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='PUT',
+            url=url,
+            auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
+            headers=headers)
+        return None
+
+    def get_bucket_acl(self, Bucket, **kwargs):
+        """获取bucket ACL"""
+        headers = mapped(kwargs)
+        url = self._conf.uri(bucket=Bucket, path="?acl")
+        logger.info("get bucket acl, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
+            headers=headers)
+        data = xml_to_dict(rt.text)
+        if data['AccessControlList'] is not None and isinstance(data['AccessControlList']['Grant'], dict):
+            lst = []
+            lst.append(data['AccessControlList']['Grant'])
+            data['AccessControlList']['Grant'] = lst
+        return data
 
     # service interface begin
     def list_buckets(self, **kwargs):
@@ -470,6 +538,10 @@ def list_buckets(self, **kwargs):
                 auth=CosS3Auth(self._conf._access_id, self._conf._access_key),
                 )
         data = xml_to_dict(rt.text)
+        if data['Buckets'] is not None and isinstance(data['Buckets']['Bucket'], dict):
+            lst = []
+            lst.append(data['Buckets']['Bucket'])
+            data['Buckets']['Bucket'] = lst
         return data
 
 if __name__ == "__main__":

qcloud_cos/test.py

@@ -147,6 +147,19 @@ def Test():
     except CosServiceError as e:
         print_error_msg(e)
 
+    print "Test Put Object ACL " + file_name
+    response = client.put_object_acl(
+        Bucket=test_bucket,
+        Key=file_name,
+        ACL='public-read-write'
+    )
+
+    print "Test Get Object ACL" + file_name
+    response = client.get_object_acl(
+        Bucket=test_bucket,
+        Key=file_name
+    )
+
     print "Test Delete Object " + file_name
     response = client.head_object(
         Bucket=test_bucket,
@@ -164,11 +177,33 @@ def Test():
         ACL='public-read'
     )
 
+    print "Test PUT Bucket ACL"
+    try:
+        response = client.put_bucket_acl(
+            Bucket='test'+file_id,
+            ACL='public-read-writea'
+        )
+    except CosServiceError as e:
+        print_error_msg(e)
+
+    print "Test GET Bucket ACL"
+    response = client.get_bucket_acl(
+        Bucket='test'+file_id,
+    )
+
     print "Test Delete Bucket"
     response = client.delete_bucket(
         Bucket='test'+file_id
     )
 
+    print "Test Head Bucket"
+    try:
+        response = client.head_bucket(
+            Bucket='test'+file_id
+        )
+    except CosServiceError as e:
+        print_error_msg(e)
+
     print "Test Create MultipartUpload"
     response = client.create_multipart_upload(
         Bucket=test_bucket,

qcloud_cos/xml2dict.py

@@ -35,12 +35,12 @@ def updateDict(self, aDict):
 
 if __name__ == "__main__":
     s = """<?xml version="1.0" encoding="utf-8" ?>
-    <result>
+    <result xmlns= "wqa.bai.com">
         <count n="1">10</count>
         <data><id>1</id><name>test1</name></data>
         <data><id>2</id><name>test2</name></data>
         <data><id>3</id><name>test3</name></data>
     </result>"""
     root = xml.etree.ElementTree.fromstring(s)
-    xmldict = XmlDictConfig(root)
+    xmldict = Xml2Dict(root)
     print xmldict