Add some high level apis

Author: dt3310321

qcloud_cos/cos_client.py

@@ -2043,6 +2043,151 @@ def append_object(self, Bucket, Key, Position, Data, **kwargs):
         response = rt.headers
         return response
 
+    def put_object_from_local_file(self, Bucket, LocalFilePath, Key, EnableMD5=False, **kwargs):
+        """本地文件上传接口，适用于小文件，最大不得超过5GB
+
+        :param Bucket(string): 存储桶名称.
+        :param LocalFilePath(string): 上传文件的本地路径.
+        :param Key(string): COS路径.
+        :param EnableMD5(bool): 是否需要SDK计算Content-MD5，打开此开关会增加上传耗时.
+        :kwargs(dict): 设置上传的headers.
+        :return(dict): 上传成功返回的结果，包含ETag等信息.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, Secret_id=secret_id, Secret_key=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 上传本地文件到cos
+            response = client.put_object(
+                Bucket='bucket',
+                LocalFilePath='local.txt',
+                Key='test.txt'
+            )
+            print response['ETag']
+        """
+        with open(LocalFilePath, 'rb') as fp:
+            return self.put_object(Bucket, fp, Key, EnableMD5, **kwargs)
+
+    def object_exists(self, Bucket, Key):
+        """判断一个文件是否存在
+
+        :param Bucket(string): 存储桶名称.
+        :param Key(string): COS路径.
+        :return(bool): 文件是否存在,返回True为存在,返回False为不存在
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, Secret_id=secret_id, Secret_key=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 上传本地文件到cos
+            status = client.object_exists(
+                Bucket='bucket',
+                Key='test.txt'
+            )
+        """
+        try:
+            self.head_object(Bucket, Key)
+            return True
+        except CosServiceError as e:
+            if e.get_status_code() == 404:
+                return False
+            else:
+                raise e
+
+    def bucket_exists(self, Bucket):
+        """判断一个存储桶是否存在
+
+        :param Bucket(string): 存储桶名称.
+        :return(bool): 存储桶
+        是否存在,返回True为存在,返回False为不存在.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, Secret_id=secret_id, Secret_key=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 上传本地文件到cos
+            status = client.bucket_exists(
+                Bucket='bucket'
+            )
+        """
+        try:
+            self.head_bucket(Bucket)
+            return True
+        except CosServiceError as e:
+            if e.get_status_code() == 404:
+                return False
+            else:
+                raise e
+
+    def change_object_storage_class(self, Bucket, Key, StorageClass):
+        """改变文件的存储类型
+
+        :param Bucket(string): 存储桶名称.
+        :param Key(string): COS路径.
+        :param StorageClass(bool): 是否需要SDK计算Content-MD5，打开此开关会增加上传耗时.
+        :kwargs(dict): 设置上传的headers.
+        :return(dict): 上传成功返回的结果，包含ETag等信息.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, Secret_id=secret_id, Secret_key=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 上传本地文件到cos
+            response = client.change_object_storage_class(
+                Bucket='bucket',
+                Key='test.txt',
+                StorageClass='STANDARD'
+            )
+        """
+        copy_source = {
+            'Bucket': Bucket,
+            'Key': Key,
+            'Region': self._conf._region,
+            'Appid': self._conf._appid
+        }
+        response = self.copy_object(
+            Bucket=Bucket,
+            Key=Key,
+            CopySource=copy_source,
+            CopyStatus='Replaced',
+            StorageClass=StorageClass
+        )
+        return response
+
+    def update_object_meta(self, Bucket, Key, **kwargs):
+        """改变文件的存储类型
+
+        :param Bucket(string): 存储桶名称.
+        :param Key(string): COS路径.
+        :kwargs(dict): 设置文件的元属性.
+        :return: None.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, Secret_id=secret_id, Secret_key=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 上传本地文件到cos
+            response = client.update_object_meta(
+                Bucket='bucket',
+                Key='test.txt',
+                ContentType='text/html'
+            )
+        """
+        copy_source = {
+            'Bucket': Bucket,
+            'Key': Key,
+            'Region': self._conf._region,
+            'Appid': self._conf._appid
+        }
+        response = self.copy_object(
+            Bucket=Bucket,
+            Key=Key,
+            CopySource=copy_source,
+            CopyStatus='Replaced',
+            **kwargs
+        )
+        return response
+
 
 if __name__ == "__main__":
     pass

qcloud_cos/streambody.py

@@ -1,5 +1,4 @@
 # -*- coding=utf-8
-import requests
 
 
 class StreamBody():

ut/test.py

@@ -671,6 +671,69 @@ def test_put_object_enable_md5():
         os.remove(file_name)
 
 
+def test_put_object_from_local_file():
+    """通过本地文件路径来上传文件"""
+    file_size = 1
+    file_id = str(random.randint(0, 1000)) + str(random.randint(0, 1000))
+    file_name = "tmp" + file_id + "_" + str(file_size) + "MB"
+    gen_file(file_name, 10)
+    with open(file_name, 'rb') as f:
+        etag = get_raw_md5(f.read())
+    put_response = client.put_object_from_local_file(
+        Bucket=test_bucket,
+        LocalFilePath=file_name,
+        Key=file_name
+    )
+    assert put_response['ETag'] == etag
+    if os.path.exists(file_name):
+        os.remove(file_name)
+
+
+def test_object_exists():
+    """测试一个文件是否存在"""
+    status = client.object_exists(
+        Bucket=test_bucket,
+        Key=test_object
+    )
+    assert status is True
+
+
+def test_bucket_exists():
+    """测试一个bucket是否存在"""
+    status = client.bucket_exists(
+        Bucket=test_bucket
+    )
+    assert status is True
+
+
+def test_change_object_storage_class():
+    """改变文件的存储类型"""
+    response = client.change_object_storage_class(
+        Bucket=test_bucket,
+        Key=test_object,
+        StorageClass='NEARLINE'
+    )
+    response = client.head_object(
+        Bucket=test_bucket,
+        Key=test_object
+    )
+    assert response['x-cos-storage-class'] == 'NEARLINE'
+
+
+def test_update_object_meta():
+    """更新文件的属性"""
+    response = client.update_object_meta(
+        Bucket=test_bucket,
+        Key=test_object,
+        ContentType='text/html'
+    )
+    response = client.head_object(
+        Bucket=test_bucket,
+        Key=test_object
+    )
+    assert response['Content-Type'] == 'text/html; charset=utf-8'
+
+
 if __name__ == "__main__":
     setUp()
     test_put_object_enable_md5()