diff --git a/qcloud_cos/cos_client.py b/qcloud_cos/cos_client.py
index 97fd5557..369583fc 100644
--- a/qcloud_cos/cos_client.py
+++ b/qcloud_cos/cos_client.py
@@ -216,6 +216,8 @@ def send_request(self, method, url, bucket, timeout=30, **kwargs):
kwargs['headers'] = format_values(kwargs['headers'])
if 'data' in kwargs:
kwargs['data'] = to_bytes(kwargs['data'])
+ if self._conf._ip is not None and self._conf._scheme == 'https':
+ kwargs['verify'] = False
for j in range(self._retry + 1):
try:
if method == 'POST':
@@ -1966,7 +1968,7 @@ def put_bucket_logging(self, Bucket, BucketLoggingStatus={}, **kwargs):
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
client = CosS3Client(config)
# 设置bucket logging服务
- logging_bucket = 'logging-beijing-1252448703'
+ logging_bucket = 'logging-beijing-1250000000'
logging_config = {
'LoggingEnabled': {
'TargetBucket': logging_bucket,
@@ -1995,7 +1997,6 @@ def put_bucket_logging(self, Bucket, BucketLoggingStatus={}, **kwargs):
auth=CosS3Auth(self._conf, params=params),
headers=headers,
params=params)
- grant_rt = self.put_bucket_acl(Bucket=Bucket, GrantFullControl=LOGGING_UIN)
return None
def get_bucket_logging(self, Bucket, **kwargs):
@@ -2110,7 +2111,7 @@ def put_bucket_domain(self, Bucket, DomainConfiguration={}, **kwargs):
"""设置bucket的自定义域名
:param Bucket(string): 存储桶名称.
- :param ReplicationConfiguration(dict): 设置Bucket的自定义域名规则.
+ :param DomainConfiguration(dict): 设置Bucket的自定义域名规则.
:param kwargs(dict): 设置请求headers.
:return: None.
@@ -2200,7 +2201,7 @@ def delete_bucket_domain(self, Bucket, **kwargs):
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
client = CosS3Client(config)
- # 获取bucket自定义域名配置
+ # 删除ucket自定义域名配置
response = client.delete_bucket_domain(
Bucket='bucket'
)
@@ -2208,7 +2209,7 @@ def delete_bucket_domain(self, Bucket, **kwargs):
headers = mapped(kwargs)
params = {'domain': ''}
url = self._conf.uri(bucket=Bucket)
- logger.info("get bucket domain, url=:{url} ,headers=:{headers}".format(
+ logger.info("delete bucket domain, url=:{url} ,headers=:{headers}".format(
url=url,
headers=headers))
rt = self.send_request(
@@ -2221,10 +2222,10 @@ def delete_bucket_domain(self, Bucket, **kwargs):
return None
def put_bucket_origin(self, Bucket, OriginConfiguration={}, **kwargs):
- """设置bucket的回源
+ """设置bucket的回源规则
:param Bucket(string): 存储桶名称.
- :param ReplicationConfiguration(dict): 设置Bucket的回源规则.
+ :param OriginConfiguration(dict): 设置Bucket的回源规则.
:param kwargs(dict): 设置请求headers.
:return: None.
@@ -2232,18 +2233,8 @@ def put_bucket_origin(self, Bucket, OriginConfiguration={}, **kwargs):
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
client = CosS3Client(config)
- # 设置bucket自定义域名配置
- origin_config = {
- 'OriginRule': [
- {
- 'OriginType': 'Redirect',
- 'OriginInfo': {
- 'HostName': 'www.abc.com',
- 'Protocol': 'HTTP'
- }
- },
- ]
- }
+ # 设置bucket回源规则
+ origin_config = {}
response = client.put_bucket_origin(
Bucket='bucket',
OriginConfiguration=origin_config
@@ -2274,13 +2265,13 @@ def get_bucket_origin(self, Bucket, **kwargs):
:param Bucket(string): 存储桶名称.
:param kwargs(dict): 设置请求headers.
- :return(dict): Bucket对应的自定义域名配置.
+ :return(dict): Bucket对应的回源规则.
.. code-block:: python
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
client = CosS3Client(config)
- # 获取bucket自定义域名配置
+ # 获取bucket回源规则
response = client.get_bucket_origin(
Bucket='bucket'
)
@@ -2313,7 +2304,7 @@ def delete_bucket_origin(self, Bucket, **kwargs):
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
client = CosS3Client(config)
- # 获取bucket自定义域名配置
+ # 删除bucket回源规则
response = client.delete_bucket_origin(
Bucket='bucket'
)
@@ -2321,7 +2312,257 @@ def delete_bucket_origin(self, Bucket, **kwargs):
headers = mapped(kwargs)
params = {'origin': ''}
url = self._conf.uri(bucket=Bucket)
- logger.info("get bucket origin, url=:{url} ,headers=:{headers}".format(
+ logger.info("delete bucket origin, url=:{url} ,headers=:{headers}".format(
+ url=url,
+ headers=headers))
+ rt = self.send_request(
+ method='DELETE',
+ url=url,
+ bucket=Bucket,
+ auth=CosS3Auth(self._conf, params=params),
+ headers=headers,
+ params=params)
+ return None
+
+ def put_bucket_inventory(self, Bucket, Id, InventoryConfiguration={}, **kwargs):
+ """设置bucket的清单规则
+
+ :param Bucket(string): 存储桶名称.
+ :param Id(string): 清单规则名称.
+ :param InventoryConfiguration(dict): Bucket的清单规则.
+ :param kwargs(dict): 设置请求headers.
+ :return: None.
+
+ .. code-block:: python
+
+ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
+ client = CosS3Client(config)
+ # 设置bucket清单规则
+ inventory_config = {
+ 'Destination': {
+ 'COSBucketDestination': {
+ 'AccountId': '100000000001',
+ 'Bucket': 'qcs::cos:ap-guangzhou::examplebucket-1250000000',
+ 'Format': 'CSV',
+ 'Prefix': 'list1',
+ 'Encryption': {
+ 'SSECOS': {}
+ }
+ },
+ 'IsEnabled': 'True',
+ 'Filter': {
+ 'Prefix': 'filterPrefix'
+ },
+ 'IncludedObjectVersions':'All',
+ 'OptionalFields': {
+ 'Field': [
+ 'Size',
+ 'LastModifiedDate',
+ 'ETag',
+ 'StorageClass',
+ 'IsMultipartUploaded',
+ 'ReplicationStatus'
+ ]
+ },
+ 'Schedule': {
+ 'Frequency': 'Daily'
+ }
+ }
+ response = client.put_bucket_inventory(
+ Bucket='bucket',
+ Id='list1',
+ InventoryConfiguration=inventory_config
+ )
+ """
+ lst = ['', ''] # 类型为list的标签
+ InventoryConfiguration['Id'] = Id
+ xml_config = format_xml(data=InventoryConfiguration, root='InventoryConfiguration', lst=lst)
+ headers = mapped(kwargs)
+ headers['Content-MD5'] = get_md5(xml_config)
+ headers['Content-Type'] = 'application/xml'
+ params = {'inventory': '', 'id': Id}
+ url = self._conf.uri(bucket=Bucket)
+ logger.info("put bucket inventory, 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_inventory(self, Bucket, Id, **kwargs):
+ """获取bucket清单规则
+
+ :param Bucket(string): 存储桶名称.
+ :param Id(string): 清单规则名称.
+ :param kwargs(dict): 设置请求headers.
+ :return(dict): Bucket对应的清单规则.
+
+ .. code-block:: python
+
+ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
+ client = CosS3Client(config)
+ # 获取bucket清单规则
+ response = client.get_bucket_inventory(
+ Bucket='bucket'
+ )
+ """
+ headers = mapped(kwargs)
+ params = {'inventory': '', 'id': Id}
+ url = self._conf.uri(bucket=Bucket)
+ logger.info("get bucket inventory, 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)
+ format_dict(data['OptionalFields'], ['Field'])
+ return data
+
+ def delete_bucket_inventory(self, Bucket, Id, **kwargs):
+ """删除bucket 回源配置
+
+ :param Bucket(string): 存储桶名称.
+ :param Id(string): 清单规则名称.
+ :param kwargs(dict): 设置请求headers.
+ :return(dict): None.
+
+ .. code-block:: python
+
+ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
+ client = CosS3Client(config)
+ # 删除bucket清单规则
+ response = client.delete_bucket_origin(
+ Bucket='bucket'
+ )
+ """
+ headers = mapped(kwargs)
+ params = {'inventory': '', 'id': Id}
+ url = self._conf.uri(bucket=Bucket)
+ logger.info("delete bucket inventory, url=:{url} ,headers=:{headers}".format(
+ url=url,
+ headers=headers))
+ rt = self.send_request(
+ method='DELETE',
+ url=url,
+ bucket=Bucket,
+ auth=CosS3Auth(self._conf, params=params),
+ headers=headers,
+ params=params)
+ return None
+
+ def put_bucket_tagging(self, Bucket, Tagging={}, **kwargs):
+ """设置bucket的标签
+
+ :param Bucket(string): 存储桶名称.
+ :param Tagging(dict): Bucket的标签集合
+ :param kwargs(dict): 设置请求headers.
+ :return: None.
+
+ .. code-block:: python
+
+ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
+ client = CosS3Client(config)
+ # 设置bucket标签
+ tagging_set = {
+ 'TagSet': {
+ 'Tag': [
+ {
+ 'Key': 'string',
+ 'Value': 'string'
+ }
+ ]
+ }
+ }
+ response = client.put_bucket_tagging(
+ Bucket='bucket',
+ Tagging=tagging_set
+ )
+ """
+ lst = ['', ''] # 类型为list的标签
+ xml_config = format_xml(data=Tagging, root='Tagging', lst=lst)
+ headers = mapped(kwargs)
+ headers['Content-MD5'] = get_md5(xml_config)
+ headers['Content-Type'] = 'application/xml'
+ params = {'tagging': ''}
+ url = self._conf.uri(bucket=Bucket)
+ logger.info("put bucket tagging, 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_tagging(self, Bucket, **kwargs):
+ """获取bucket标签
+
+ :param Bucket(string): 存储桶名称.
+ :param kwargs(dict): 设置请求headers.
+ :return(dict): Bucket对应的标签.
+
+ .. code-block:: python
+
+ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
+ client = CosS3Client(config)
+ # 获取bucket标签
+ response = client.get_bucket_tagging(
+ Bucket='bucket'
+ )
+ """
+ headers = mapped(kwargs)
+ params = {'tagging': ''}
+ url = self._conf.uri(bucket=Bucket)
+ logger.info("get bucket tagging, 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)
+ if 'TagSet' in data:
+ format_dict(data['TagSet'], ['Tag'])
+ return data
+
+ def delete_bucket_tagging(self, Bucket, **kwargs):
+ """删除bucket 回源配置
+
+ :param Bucket(string): 存储桶名称.
+ :param kwargs(dict): 设置请求headers.
+ :return(dict): None.
+
+ .. code-block:: python
+
+ config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token) # 获取配置对象
+ client = CosS3Client(config)
+ # 删除bucket标签
+ response = client.delete_bucket_tagging(
+ Bucket='bucket'
+ )
+ """
+ headers = mapped(kwargs)
+ params = {'tagging': ''}
+ url = self._conf.uri(bucket=Bucket)
+ logger.info("delete bucket tagging, url=:{url} ,headers=:{headers}".format(
url=url,
headers=headers))
rt = self.send_request(
diff --git a/qcloud_cos/cos_comm.py b/qcloud_cos/cos_comm.py
index 6e48353b..9a69817f 100644
--- a/qcloud_cos/cos_comm.py
+++ b/qcloud_cos/cos_comm.py
@@ -18,7 +18,6 @@
SINGLE_UPLOAD_LENGTH = 5*1024*1024*1024 # 单次上传文件最大为5GB
DEFAULT_CHUNK_SIZE = 1024*1024 # 计算MD5值时,文件单次读取的块大小为1MB
-LOGGING_UIN = 'id="qcs::cam::uin/100001001014:uin/100001001014"'
# kwargs中params到http headers的映射
maplist = {
'ContentLength': 'Content-Length',
@@ -377,7 +376,7 @@ def format_dict(data, key_lst):
return data
for key in key_lst:
# 将dict转为list,保持一致
- if key in data and isinstance(data[key], dict):
+ if key in data and (isinstance(data[key], dict) or isinstance(data[key], str)):
lst = []
lst.append(data[key])
data[key] = lst
diff --git a/qcloud_cos/version.py b/qcloud_cos/version.py
index f90059ab..9ec878f7 100644
--- a/qcloud_cos/version.py
+++ b/qcloud_cos/version.py
@@ -1,2 +1,2 @@
-__version__ = '5.1.7.5'
+__version__ = '5.1.7.6'
diff --git a/setup.py b/setup.py
index 91e324ac..008d06a8 100644
--- a/setup.py
+++ b/setup.py
@@ -16,7 +16,7 @@ def long_description():
setup(
name='cos-python-sdk-v5',
- version='1.7.5',
+ version='1.7.6',
url='https://www.qcloud.com/',
license='MIT',
author='tiedu, lewzylu, channingliu',
diff --git a/ut/test.py b/ut/test.py
index 32cc1c2b..d06f55c2 100644
--- a/ut/test.py
+++ b/ut/test.py
@@ -831,20 +831,6 @@ def test_bucket_exists():
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_put_get_bucket_policy():
"""设置获取bucket的policy配置"""
resource = "qcs::cos:" + REGION + ":uid/" + APPID + ":" + test_bucket + "/*"
@@ -945,19 +931,90 @@ def test_put_get_delete_bucket_domain():
)
-def test_put_get_delete_bucket_origin():
- """测试设置获取删除bucket回源域名"""
- origin_config = {
- 'OriginRule': [
- {
- 'OriginType': 'Redirect',
- 'OriginInfo': {
- 'HostName': 'www.abc.com',
- 'Protocol': 'HTTP'
+def test_put_get_delete_bucket_inventory():
+ """测试设置获取删除bucket清单"""
+ inventory_config = {
+ 'Destination': {
+ 'COSBucketDestination': {
+ 'AccountId': '2779643970',
+ 'Bucket': 'qcs::cos:' + REGION + '::' + test_bucket,
+ 'Format': 'CSV',
+ 'Prefix': 'list1',
+ 'Encryption': {
+ 'SSECOS': {}
}
- },
- ]
+ }
+ },
+ 'IsEnabled': 'True',
+ 'Filter': {
+ 'Prefix': 'filterPrefix'
+ },
+ 'IncludedObjectVersions': 'All',
+ 'OptionalFields': {
+ 'Field': [
+ 'Size',
+ 'LastModifiedDate',
+ 'ETag',
+ 'StorageClass',
+ 'IsMultipartUploaded',
+ 'ReplicationStatus'
+ ]
+ },
+ 'Schedule': {
+ 'Frequency': 'Daily'
+ }
+ }
+ response = client.put_bucket_inventory(
+ Bucket=test_bucket,
+ Id='test',
+ InventoryConfiguration=inventory_config
+ )
+ # wait for sync
+ # get inventory
+ time.sleep(4)
+ response = client.get_bucket_inventory(
+ Bucket=test_bucket,
+ Id='test'
+ )
+ # delete inventory
+ response = client.delete_bucket_inventory(
+ Bucket=test_bucket,
+ Id='test'
+ )
+
+
+def test_put_get_delete_bucket_tagging():
+ """测试设置获取删除bucket标签"""
+ tagging_config = {
+ 'TagSet': {
+ 'Tag': [
+ {
+ 'Key': 'key0',
+ 'Value': 'value0'
+ }
+ ]
+ }
}
+ response = client.put_bucket_tagging(
+ Bucket=test_bucket,
+ Tagging=tagging_config
+ )
+ # wait for sync
+ # get tagging
+ time.sleep(1)
+ response = client.get_bucket_tagging(
+ Bucket=test_bucket
+ )
+ assert tagging_config == response
+ # delete tagging
+ response = client.delete_bucket_tagging(
+ Bucket=test_bucket
+ )
+
+
+def _test_put_get_delete_bucket_origin():
+ """测试设置获取删除bucket回源域名"""
+ origin_config = {}
response = client.put_bucket_origin(
Bucket=test_bucket,
OriginConfiguration=origin_config
@@ -976,6 +1033,7 @@ def test_put_get_delete_bucket_origin():
if __name__ == "__main__":
setUp()
+ """
test_put_object_enable_md5()
test_upload_with_server_side_encryption()
test_upload_empty_file()
@@ -993,4 +1051,6 @@ def test_put_get_delete_bucket_origin():
test_put_get_bucket_policy()
test_put_file_like_object()
test_put_chunked_object()
+ """
+ test_put_get_delete_bucket_inventory()
tearDown()