release(1.9.40): feat/symlink

Author: libertyzhu

qcloud_cos/cos_client.py

@@ -4576,6 +4576,86 @@ def update_object_meta(self, Bucket, Key, **kwargs):
         )
         return response
 
+    def put_symlink(self, Bucket, SymlinkName, SymlinkTarget, **kwargs):
+        """创建软链接
+
+        :param Bucket(string): 存储桶名称.
+        :param SymlinkName(string): 软链接路径.
+        :param SymlinkTarget(string): 目标路径.
+        :kwargs(dict): 公共请求头部.
+        :return: 请求的响应头部.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 创建软链接
+            response = client.put_symlink(
+                Bucket='bucket-1250000000',
+                SymlinkName='testsym.txt',
+                SymlinkTarget='test.txt',
+            )
+        """
+
+        headers = mapped(kwargs)
+        headers['x-cos-symlink-target'] = SymlinkTarget
+        params = {'symlink': ''}
+
+        url = self._conf.uri(bucket=Bucket, path=SymlinkName)
+        logger.debug("put symlink, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='PUT',
+            url=url,
+            bucket=Bucket,
+            auth=CosS3Auth(self._conf, SymlinkName, params=params),
+            headers=headers,
+            params=params)
+        response = dict(**rt.headers)
+        return response
+
+    def get_symlink(self, Bucket, SymlinkName, **kwargs):
+        """获取软链接本身
+
+        :param Bucket(string): 存储桶名称.
+        :param SymlinkName(string): 软链接路径.
+        :kwargs(dict): 公共请求头部.
+        :return: 请求的响应头部.
+
+        .. code-block:: python
+
+            config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token)  # 获取配置对象
+            client = CosS3Client(config)
+            # 获取软链接本身
+            response = client.get_symlink(
+                Bucket='bucket-1250000000',
+                SymlinkName='testsym.txt',
+            )
+        """
+
+        headers = mapped(kwargs)
+        params = {'symlink': ''}
+
+        if 'versionId' in headers:
+            params['versionId'] = headers['versionId']
+            del headers['versionId']
+        params = format_values(params)
+
+        url = self._conf.uri(bucket=Bucket, path=SymlinkName)
+        logger.debug("get symlink, url=:{url} ,headers=:{headers}".format(
+            url=url,
+            headers=headers))
+        rt = self.send_request(
+            method='GET',
+            url=url,
+            bucket=Bucket,
+            auth=CosS3Auth(self._conf, SymlinkName, params=params),
+            headers=headers,
+            params=params)
+        response = dict(**rt.headers)
+        return response
+
     def put_bucket_encryption(self, Bucket, ServerSideEncryptionConfiguration={}, **kwargs):
         """设置执行存储桶下的默认加密配置
 

qcloud_cos/version.py

@@ -1 +1 @@
-__version__ = '5.1.9.39'
+__version__ = '5.1.9.40'

setup.py

@@ -33,7 +33,7 @@ def long_description():
 
 setup(
     name='cos_python_sdk_v5', # comply with PEP 625
-    version='1.9.39',
+    version='1.9.40',
     url='https://www.qcloud.com/',
     license='MIT',
     author='tiedu, lewzylu, channingliu',

ut/test.py

@@ -6750,6 +6750,83 @@ def test_put_object_with_tagging():
         os.remove(filename)
 
 
+def test_put_get_symlink_multiver():
+    """测试创建和获取软链接功能"""
+    # 先上传一个目标文件
+    target_file_name = "test_symlink_target.txt"
+    symlink_name = "test_symlink.txt"
+
+    # 开启版本控制 测试带版本控制的软链接功能
+    response = client.put_bucket_versioning(
+        Bucket=test_bucket,
+        Status='Enabled',
+    )
+    
+    # 上传目标文件
+    response = client.put_object(
+        Bucket=test_bucket,
+        Body=b'hello',
+        Key=target_file_name
+    )
+    
+    # 创建软链接，指向目标文件
+    response = client.put_symlink(
+        Bucket=test_bucket,
+        SymlinkName=symlink_name,
+        SymlinkTarget=target_file_name
+    )
+    print(response)
+    
+    # 验证put_symlink响应
+    assert 'x-cos-request-id' in response
+    assert 'x-cos-version-id' in response
+    version_id = response['x-cos-version-id']
+    
+    # 获取软链接信息
+    response = client.get_symlink(
+        Bucket=test_bucket,
+        SymlinkName=symlink_name,
+        VersionId=version_id,
+    )
+    print(response)
+    
+    # 验证get_symlink响应
+    assert 'x-cos-request-id' in response
+    assert 'x-cos-symlink-target' in response
+    assert response['x-cos-symlink-target'] == target_file_name
+
+    # 读取软链接指向文件
+    response = client.get_object(
+        Bucket=test_bucket,
+        Key=symlink_name,
+    )
+    body = response['Body'].read()
+    print(response)
+    print(body)
+    assert body == b'hello'
+
+    # 获取软链接信息 -- 使用无效的verionId
+    try:
+        client.get_symlink(
+            Bucket=test_bucket,
+            SymlinkName=symlink_name,
+            VersionId='invalid-version-id',
+        )
+    except CosServiceError as e:
+        assert e.get_error_code() == 'NoSuchVersion'
+
+    # 获取软链接信息 -- 使用无效的symlinkName
+    try:
+        client.get_symlink(
+            Bucket=test_bucket,
+            SymlinkName='invalid-sym',
+        )
+    except CosServiceError as e:
+        assert e.get_error_code() == 'NoSuchKey'
+
+    # 清理测试文件
+    client.delete_object(Bucket=test_bucket, Key=symlink_name)
+    client.delete_object(Bucket=test_bucket, Key=target_file_name)
 
 if __name__ == "__main__":
     setUp()