### Install KS3 SDK for Python (Local) Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/README.md Clone the repository and install the SDK locally. ```bash git clone https://gitee.com/ks3sdk/ks3-python-sdk.git cd ks3-python-sdk python setup.py install ``` -------------------------------- ### Install KS3 SDK for Python (Online) Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/README.md Use pip for online installation of the KS3 SDK for Python. ```bash pip install ks3sdk ``` -------------------------------- ### Run KS3 Python SDK Examples Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Execute example scripts located in the examples/ directory. Ensure necessary environment variables like access keys and bucket names are set before running. ```bash export KS3_IAM_ACCESS_KEY_ID="your_access_key" export KS3_IAM_ACCESS_KEY_SECRET="your_secret_key" export KS3_BUCKET="your_test_bucket" python examples/basic.py ``` -------------------------------- ### Install KS3 Python SDK Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Install the SDK in development mode or as a regular package. Dependencies are managed via setup.py. ```bash pip install -e . python setup.py install ``` -------------------------------- ### List Objects with Prefix and Time Range Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists objects within a bucket matching a prefix and within a specified time range (start and end times). ```python # startTime = 1625460400 这个时间之前的数据 # endTime = 1625460400 这个时间之后的数据 # startTime = 1525460400 ,endTime=1625460400 这个时间中间的数据 keys = b.listObjects(filename="/Users/xxx/a.txt", prefix="local/load", endTime=1625460400) ``` -------------------------------- ### PUT and GET with Client-Side Encryption - Python Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/ENCRYPTION.md Perform PUT and GET operations using an encrypted KS3 connection. The usage is similar to a standard client, with encryption parameters set during initialization. ```python from ks3.connection import Connection ak = 'YOUR_ACCESS_KEY' sk = 'YOUR_SECRET_KEY' c = Connection('', '', host='',is_secure=False, domain_mode=False, local_encrypt=True, local_key_path="your_key_path") b = c.get_bucket("your_bucket") #put kw = b.new_key("your_key") ret = kw.set_contents_from_string("some string") #get get_key = b.get_key("your_key") s = get_key.get_contents_as_string() print("Result:",s) ``` -------------------------------- ### Set Bucket Policy Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Configures bucket policies for access control. Supports setting, getting, and deleting policies using a JSON-formatted string. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) bucket = c.get_bucket('') # 设置空间策略 bucket.set_bucket_policy( policy='{"Statement":[{"Resource":["krn:ksc:ks3:::jiangran123","krn:ksc:ks3:::jiangran123/*"],"Principal":{"KSC":["krn:ksc:iam::32432423:root"]},"Action":["ks3:*"],"Effect":"Allow"}]}') # 获取空间策略 # policy = bucket.get_bucket_policy() # 删除空间策略 # bucket.delete_bucket_policy() ``` -------------------------------- ### Multipart Upload of Large Files Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Perform a multipart upload for large files by breaking them into smaller chunks. This requires installing the 'FileChunkIO' library (`pip install FileChunkIO`). ```python import math, os from ks3.connection import Connection from filechunkio import FileChunkIO c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') source_path = 'SOURCE_FILE_PATH' source_size = os.stat(source_path).st_size # 获取初始化的 uploadId,之后的操作中将会用到 # 此处 os.path.basename(source_path) 可以替换为需要设置的 objectKey mp = b.initiate_multipart_upload(os.path.basename(source_path)) # 举例以 50 MiB 为分块大小 chunk_size = 52428800 chunk_count = int(math.ceil(source_size*1.0 / chunk_size*1.0)) # 通过 FileChunkIO 将文件分片 try: for i in range(chunk_count): offset = chunk_size * i bytes = min(chunk_size, source_size - offset) with FileChunkIO(source_path, 'r', offset=offset, bytes=bytes) as fp: mp.upload_part_from_file(fp, part_num=i + 1) # 发送请求,合并分片,完成分片上传 ret = mp.complete_upload() if ret and ret.status == 200: print("上传成功") except: pass ``` -------------------------------- ### Get Object Access Permissions Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Retrieves the access control list (ACL) for an object. ```APIDOC ## Get Object Access Permissions ### Description Retrieves the access control list (ACL) for a specific object. ### Method GET (Implicit) ### Endpoint `//` ### Code Example ```python # Assuming 'k' is the Key object # acl = k.get_acl() # print(acl) ``` ``` -------------------------------- ### Get Object Metadata Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Retrieves metadata for an object, such as size and last modified time. ```APIDOC ## Get Object Metadata ### Description Retrieves metadata for an object, including its size and last modified timestamp. ### Method GET (Implicit) ### Endpoint `//` ### Code Example ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') k = b.get_key('') if k: print(k.name, k.size, k.last_modified) ``` ``` -------------------------------- ### Get Object Access Permissions Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Retrieves the Access Control List (ACL) for a specific object in a bucket. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') policy = b.get_acl('') print(policy.to_xml()) ``` -------------------------------- ### Get Object Access Permissions Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Retrieves the Access Control List (ACL) for a specific object. ```APIDOC ## Get Object Access Permissions ### Description Retrieves the Access Control List (ACL) for a specific object. ### Method ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') policy = b.get_acl('') print(policy.to_xml()) ``` ``` -------------------------------- ### Get Object Metadata Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Retrieves metadata for an object, including its size and last modified timestamp. Requires the bucket name and object key. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') k = b.get_key('') if k: print(k.name, k.size, k.last_modified) ``` -------------------------------- ### Get and Set Bucket ACL Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Manages bucket access control lists (ACLs). Allows retrieving the current ACL and setting it to predefined values like 'public-read'. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') # 获取 Bucket 访问权限,并打印 policy = b.get_acl() for grant in policy.acl.grants: print(grant.permission, grant.display_name, grant.email_address, grant.id) # 设置bucket的权限, private or public-read or public-read-write b.set_acl("public-read") ``` -------------------------------- ### Initialize KS3 Connection Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Establish a connection to the KS3 service using your access key, secret key, and the region endpoint. This is the primary entry point for SDK operations. ```python from ks3.connection import Connection conn = Connection(access_key, secret_key, host='region_endpoint') ``` -------------------------------- ### Initialize Encrypted Connection - Python Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/ENCRYPTION.md Initialize a KS3 connection with client-side encryption enabled. Set `local_encrypt` to true and provide the `local_key_path` to your key file. ```python from ks3.connection import Connection ak = 'YOUR_ACCESS_KEY' sk = 'YOUR_SECRET_KEY' c = Connection('', '', host='',is_secure=False, domain_mode=False, local_encrypt=True, local_key_path="your_key_path") ``` -------------------------------- ### Download KS3 Object to File Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md Download an object from a bucket and save it to a specified file path on your local system. ```python from ks3.connection import Connection c = Connection('', '', host='YOUR_REGION_ENDPOINT') key_name = "YOUR_KEY_NAME" b = c.get_bucket("") k = b.get_key(key_name) k.get_contents_to_filename("") ``` -------------------------------- ### Initialize KS3 Connection Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/README.md Initialize the KS3 connection with access key, secret key, and region endpoint. Ensure to replace placeholders with your actual credentials and endpoint. The `is_secure` and `domain_mode` parameters control the connection protocol and custom domain usage, respectively. `timeout` sets the connection timeout in seconds. ```python from ks3.connection import Connection ak = 'YOUR_ACCESS_KEY' sk = 'YOUR_SECRET_KEY' c = Connection(ak, sk, host='YOUR_REGION_ENDPOINT', is_secure=False, domain_mode=False,timeout = 1) ``` -------------------------------- ### Upload with Callback URL and Body Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Configure upload callbacks by setting the 'x-kss-callbackurl' and 'x-kss-callbackbody' headers. Refer to the documentation for detailed callback handling. ```python headers = {"x-kss-callbackurl": "Callback_URL", "x-kss-callbackbody":"objectKey=${key}&etag=${etag}&uid=123"} ret = k.set_contents_from_filename("", headers=headers) ``` -------------------------------- ### Fetch Network Resource and Upload as Object Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Upload an object by fetching it from a specified network URL. The 'x-kss-acl' header can be set to 'public-read' for public access; otherwise, it defaults to private. ```python bucket = conn.get_bucket('') bucket.fetch_object('www-logo', source_url='http://fe.ksyun.com/project/resource/img/www-logo.png', headers={'x-kss-acl': 'public-read'}) ``` -------------------------------- ### Create a KS3 Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md Use this snippet to create a new bucket. Bucket names must be globally unique. A 409 conflict error indicates a name collision. ```python from ks3.connection import Connection c = Connection('', '', host='YOUR_REGION_ENDPOINT') # 这里如果出现409 conflict错误,说明请求的bucket name有冲突,因为bucket name是全局唯一的 b = c.create_bucket("") ``` -------------------------------- ### Create Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Creates a new bucket to store key/value data. ```APIDOC ## Create Bucket ### Description Creates a new bucket. Buckets serve as containers for storing key/value data. ### Method PUT (Implicit) ### Endpoint `/` ### Notes If a 409 conflict error occurs, it indicates a naming conflict as Bucket names are globally unique. ### Code Example ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.create_bucket('') ``` ``` -------------------------------- ### List Buckets Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists all buckets associated with the customer's account. Requires establishing a connection with access keys and region endpoint. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) buckets = c.get_all_buckets() for b in buckets: print(b.name) ``` -------------------------------- ### Run Unit Tests for KS3 Python SDK Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Execute unit tests using the unittest framework. Tests are organized by module in the unittests/ directory. You can run all tests, specific files, or individual test cases. ```bash python -m unittest discover unittests python -m unittest unittests.test_bucket -v python -m unittest unittests.test_bucket.TestBucket.test_complete_multipart_upload_with_parts_list -v ``` -------------------------------- ### Complete Multipart Upload - Using xml_body (Traditional Method) Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md This method demonstrates the traditional way of completing a multipart upload by manually constructing an XML string for the `xml_body` parameter. ```APIDOC ## Complete Multipart Upload - Using xml_body (Traditional Method) ### Description This method demonstrates the traditional way of completing a multipart upload by manually constructing an XML string for the `xml_body` parameter. ### Method `complete_multipart_upload` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **key_name** (str) - Required - The object key name. - **upload_id** (str) - Required - The Upload ID for the multipart upload. - **xml_body** (str) - Required - Manually constructed XML string representing the completed parts. - **headers** (dict) - Optional - Custom request headers. ### Request Example ```python from ks3.connection import Connection conn = Connection('', '', host='ks3-cn-beijing.ksyuncs.com') b = conn.get_bucket('') # Manually construct XML xml_body = '\n' for part in result.data: xml_body += ' \n' xml_body += ' %d\n' % part.part_number xml_body += ' %s\n' % part.etag xml_body += ' \n' xml_body += '' # Complete the multipart upload ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', xml_body=xml_body ) ``` ### Response #### Success Response (200) Details of the completed multipart upload. #### Response Example ```json { "ETag": "\"some-etag\"", "Location": "http://bucket.ks3.com/object-key" } ``` ``` -------------------------------- ### Download Object as String Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Downloads the content of an object and returns it as a decoded string. Includes basic exception handling for missing objects. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') try: k = b.get_key('') s = k.get_contents_as_string().decode() print(s) except: pass # 异常处理 ``` -------------------------------- ### Write Unit Tests with unittest.mock Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Implement unit tests using Python's built-in unittest framework. Utilize `unittest.mock.Mock` and `patch` for mocking external dependencies, testing both normal and error scenarios. ```python # unittests/test_xxx.py import unittest from unittest.mock import Mock, patch class TestYourFeature(unittest.TestCase): def setUp(self): # 测试前的准备工作 pass def test_basic_functionality(self): # 测试基本功能 pass def test_error_handling(self): # 测试错误处理 pass ``` -------------------------------- ### List Objects in a Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists all files and directories within a bucket. Differentiates between files (Key) and directories (Prefix). ```python from ks3.connection import Connection from ks3.prefix import Prefix from ks3.key import Key c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') keys = b.list() for k in keys: if isinstance(k, Key): print('file:%s' % k.name) elif isinstance(k, Prefix) print('dir:%s' % k.name) ``` -------------------------------- ### List Objects in KS3 Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md List all objects and directories within a specified bucket. The output includes objects and prefixes. ```python from ks3.connection import Connection from ks3.prefix import Prefix from ks3.key import Key c = Connection('', '', host='YOUR_REGION_ENDPOINT') b = c.get_bucket("") keys = b.list() for k in keys: print('object:', k) ``` -------------------------------- ### List Objects with Prefix Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists objects within a bucket that match a specific prefix. ```python keys = b.list(prefix="PREFIX") ``` -------------------------------- ### Initiate Multipart Upload with Infrequent Access Storage Class Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md When initiating a multipart upload, you can specify the storage class, such as 'STANDARD_IA' for infrequent access, using the 'x-kss-storage-class' header. ```python # x-kss-storage-class有效值为"STANDARD"、"STANDARD_IA"。"STANDARD"表示标准存储,"STANDARD_IA"表示低频存储,如果不指定,默认为标准存储。 headers = {"x-kss-storage-class": "STANDARD_IA"} mp = b.initiate_multipart_upload('', policy="private", headers=headers) ``` -------------------------------- ### List Buckets Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists all buckets associated with the customer's account. ```APIDOC ## List Buckets ### Description Lists all buckets associated with the customer's account. ### Method GET (Implicit) ### Endpoint / ### Code Example ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) buckets = c.get_all_buckets() for b in buckets: print(b.name) ``` ``` -------------------------------- ### Create Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Creates a new bucket, which serves as a container for key/value data in S3. Bucket names must be globally unique. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.create_bucket('') ``` -------------------------------- ### Complete Multipart Upload - Using parts parameter (Dictionary List) Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md This method uses the `parts` parameter with a list of dictionaries, where each dictionary represents a part and must contain 'part_number' and 'etag' (or their uppercase equivalents). ```APIDOC ## Complete Multipart Upload - Using parts parameter (Dictionary List) ### Description This method uses the `parts` parameter with a list of dictionaries, where each dictionary represents a part and must contain 'part_number' and 'etag' (or their uppercase equivalents). This provides flexibility in how part information is supplied. ### Method `complete_multipart_upload` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **key_name** (str) - Required - The object key name. - **upload_id** (str) - Required - The Upload ID for the multipart upload. - **parts** (list[dict]) - Required - A list of dictionaries, where each dictionary must have `part_number`/`PartNumber` and `etag`/`ETag` keys. - **headers** (dict) - Optional - Custom request headers. ### Request Example ```python # Using a list of dictionaries parts = [ {'part_number': 1, 'etag': '"abc123"'}, {'part_number': 2, 'etag': '"def456"'}, {'part_number': 3, 'etag': '"ghi789"'} ] ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=parts ) # Using uppercase keys is also supported: parts_uppercase = [ {'PartNumber': 1, 'ETag': '"abc123"'}, {'PartNumber': 2, 'ETag': '"def456"'} ] ret_uppercase = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=parts_uppercase ) ``` ### Response #### Success Response (200) Details of the completed multipart upload. #### Response Example ```json { "ETag": "\"some-etag\"", "Location": "http://bucket.ks3.com/object-key" } ``` ``` -------------------------------- ### Download Object to File Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Saves the content of an object directly to a specified file path on the local filesystem. ```python #保存到文件 k.get_contents_to_filename("SAVED_FILE_PATH") ``` -------------------------------- ### Perform Bucket Operations with KS3 SDK Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Interact with KS3 buckets, including retrieving a bucket object, listing its contents, and creating new buckets. Requires an initialized Connection object. ```python bucket = conn.get_bucket('bucket_name') # 列出对象 keys = bucket.list() # 创建新存储桶 bucket = conn.create_bucket('new_bucket_name') ``` -------------------------------- ### Error Handling for Complete Multipart Upload - Missing Parameters Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md Demonstrates error handling when neither 'xml_body' nor 'parts' parameters are provided to the complete_multipart_upload method. A ValueError is raised. ```python # 如果两个参数都不提供 try: ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id' ) except ValueError as e: print(f"错误:{e}") # 输出:Either xml_body or parts parameter must be provided ``` -------------------------------- ### List Bucket Contents Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists files or directories within a bucket. ```APIDOC ## List Bucket Contents ### Description Lists files or directories within a specified bucket. ### Method GET (Implicit) ### Endpoint `/` ### Code Example ```python # Assuming 'b' is the Bucket object # for key_info in b.list(): # print(key_info.name) ``` ``` -------------------------------- ### Download KS3 Object as String Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md Retrieve the content of a KS3 object as a string. The content is decoded from bytes to a string. ```python #打印object内容 s = k.get_contents_as_string().decode() print(s) ``` -------------------------------- ### Error Handling for Complete Multipart Upload - Both Parameters Provided Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md Illustrates error handling when both 'xml_body' and 'parts' parameters are simultaneously provided to the complete_multipart_upload method. A ValueError is raised. ```python # 如果同时提供两个参数 try: ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', xml_body='', parts=[...] ) except ValueError as e: print(f"错误:{e}") # 输出:Cannot specify both xml_body and parts parameters ``` -------------------------------- ### List All In-Progress Multipart Upload Tasks in a Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md List all multipart upload tasks that are currently in progress within a KS3 bucket. This includes details for each uploaded part. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') # 列举 Bucket 中所有的已上传分片信息 for p in b.list_multipart_uploads(): print('uploadId:%s,key:%s' % (p.id, p.key_name)) for i in p: print(i.part_number, i.size, i.etag, i.last_modified) ``` -------------------------------- ### Complete Multipart Upload - Using parts parameter (Part Object List) Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md This method utilizes the `parts` parameter by providing a list of `Part` objects, typically obtained from the `data` attribute of a `list_parts` response. ```APIDOC ## Complete Multipart Upload - Using parts parameter (Part Object List) ### Description This method utilizes the `parts` parameter by providing a list of `Part` objects, typically obtained from the `data` attribute of a `list_parts` response. This offers a structured way to represent the uploaded parts. ### Method `complete_multipart_upload` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **key_name** (str) - Required - The object key name. - **upload_id** (str) - Required - The Upload ID for the multipart upload. - **parts** (list[Part]) - Required - A list of `Part` objects, each containing `part_number` and `etag`. - **headers** (dict) - Optional - Custom request headers. ### Request Example ```python from ks3.multipart import Part # Get Part object list from list_parts result = b.list_parts( bucket_name=b.name, key_name='object-key', upload_id='upload-id' ) # Use the list of Part objects ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=result.data # result.data is a list of Part objects ) ``` ### Response #### Success Response (200) Details of the completed multipart upload. #### Response Example ```json { "ETag": "\"some-etag\"", "Location": "http://bucket.ks3.com/object-key" } ``` ``` -------------------------------- ### Generate Encryption Key - Python Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/ENCRYPTION.md Use this method to generate a new encryption key and save it to a specified path. Ensure the key name is descriptive. ```python from ks3.encryption import Crypts Crypts.generate_key('your_path', 'key_name') ``` -------------------------------- ### Perform Object Operations with KS3 SDK Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Manage objects within a KS3 bucket. Upload from a local file or download to a local file. Requires a Bucket object. ```python # 从文件上传 key = bucket.new_key('object_name') key.set_contents_from_filename('local_file_path') # 下载到文件 key = bucket.get_key('object_name') key.get_contents_to_filename('local_file_path') ``` -------------------------------- ### Bucket Policy Management Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Configures bucket policies for access control. ```APIDOC ## Bucket Policy Management ### Description Configures bucket policies for access control, supporting addition, query, and deletion of policies. ### Method PUT/GET/DELETE (Implicit) ### Endpoint `/` ### Code Example (Set Policy) ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) bucket = c.get_bucket('') bucket.set_bucket_policy( policy='{"Statement":[{"Resource":["krn:ksc:ks3:::jiangran123","krn:ksc:ks3:::jiangran123/*"],"Principal":{"KSC":["krn:ksc:iam::32432423:root"]},"Action":["ks3:*"],"Effect":"Allow"}]}') ``` ### Code Example (Get Policy - commented out) ```python # policy = bucket.get_bucket_policy() ``` ### Code Example (Delete Policy - commented out) ```python # bucket.delete_bucket_policy() ``` ### References - [Bucket Policy](https://docs.ksyun.com/documents/5177) - [Permissions Overview](https://docs.ksyun.com/documents/5170) ``` -------------------------------- ### Multipart Upload with FileChunkIO (Encrypted) - Python Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/ENCRYPTION.md Perform a multipart upload using `FileChunkIO` with client-side encryption. Ensure the KS3 connection is initialized with encryption parameters. ```python c = Connection('', '', host,is_secure=False, domain_mode=False, local_encrypt=True, local_key_path="your_key_path") #continue normal multipart upload method ``` -------------------------------- ### List All KS3 Buckets Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md Retrieve a list of all buckets associated with your account. The code iterates through the buckets and prints their names. ```python from ks3.connection import Connection c = Connection('', '', host='YOUR_REGION_ENDPOINT') buckets = c.get_all_buckets() for b in buckets: print(b.name) ``` -------------------------------- ### Upload File to KS3 Bucket Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md Upload a file from a local path to a KS3 bucket. This method also allows specifying the file's ACL. ```python from ks3.connection import Connection c = Connection('', '', host='YOUR_REGION_ENDPOINT') b = c.get_bucket("") k = b.new_key("") ret=k.set_contents_from_filename("") if ret and ret.status == 200: print("上传成功") ``` -------------------------------- ### Generate Download URL Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Generates a pre-signed URL for downloading a private object. The URL has a specified expiration time. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') k = b.get_key('') if k: url = k.generate_url(60) # 60s 后该链接过期 print(url) ``` ```python k.generate_url(1492073594, expires_in_absolute=True) # 1492073594为Unix Time ``` -------------------------------- ### Generate Download URL Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Generates a pre-signed URL for downloading an object. ```APIDOC ## Generate Download URL ### Description Generates a pre-signed URL that allows temporary access for downloading an object. ### Method GET (Implicit, for URL generation) ### Endpoint `//` ### Code Example ```python # Assuming 'k' is the Key object # download_url = k.generate_url(expires_in=3600) # URL valid for 1 hour # print(download_url) ``` ``` -------------------------------- ### Upload Object with Infrequent Access Storage Class Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Upload an object using the Infrequent Access (STANDARD_IA) storage class. This is specified via the 'x-kss-storage-class' header. The default is STANDARD. ```python # x-kss-storage-class有效值为"STANDARD"、"STANDARD_IA"。"STANDARD"表示标准存储,"STANDARD_IA"表示低频存储,如果不指定,默认为标准存储。 headers = {"x-kss-storage-class": "STANDARD_IA"} resp = k.set_contents_from_filename("", policy="private", headers=headers) ``` -------------------------------- ### List Bucket Contents Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists files and directories within a bucket. Supports filtering by prefix and grouping by delimiter. ```APIDOC ## List Bucket Contents ### Description Lists files and directories within a bucket. Supports filtering by prefix and grouping by delimiter. ### Method ```python from ks3.connection import Connection from ks3.prefix import Prefix from ks3.key import Key c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') keys = b.list() for k in keys: if isinstance(k, Key): print('file:%s' % k.name) elif isinstance(k, Prefix): print('dir:%s' % k.name) ``` ### List Objects with Prefix ```python keys = b.list(prefix="PREFIX") ``` ### List Objects with Delimiter ```python # delimiter is the character used to group file names keys = b.list(delimiter='/') # If delimiter is empty, it defaults to '/' # Characters before the delimiter in the result will be placed in commonPrefixes, similar to understanding folders ``` ### List Objects with Prefix and Time Range ```python # startTime = data before this time 1625460400 # endTime = data after this time 1625460400 # startTime = data between this time 1525460400, endTime=1625460400 keys = b.listObjects(filename="/Users/xxx/a.txt", prefix="local/load", endTime=1625460400) ``` ``` -------------------------------- ### List Objects with Delimiter Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Lists objects in a bucket, grouping them by a specified delimiter. Common prefixes are collected, similar to folder structures. ```python # delimiter 为对文件名称进行分组的字符 keys = b.list(delimiter='/') # delimiter为空时,默认为'/' # 返回结果中 delimiter 分隔符之前的字符会放入 commonPrefixes 中,可以类比理解为文件夹 ``` -------------------------------- ### Generate Download URL Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Generates a pre-signed URL for downloading a private object. The URL has a limited validity period. ```APIDOC ## Generate Download URL ### Description Generates a pre-signed URL for downloading a private object. The URL has a limited validity period. ### Method ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') k = b.get_key('') if k: url = k.generate_url(60) # URL expires after 60s print(url) ``` ### Generate URL with Absolute Expiration Timestamp ```python k.generate_url(1492073594, expires_in_absolute=True) # 1492073594 is the Unix Time ``` ``` -------------------------------- ### Complete Multipart Upload with Dictionary List Parts Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md This method allows completing a multipart upload using a list of dictionaries, where each dictionary represents a part and must contain 'part_number' and 'etag' (or their uppercase equivalents). ```python # 使用字典列表 parts = [ {'part_number': 1, 'etag': '"abc123"'}, {'part_number': 2, 'etag': '"def456"'}, {'part_number': 3, 'etag': '"ghi789"'} ] ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=parts ) ``` ```python parts = [ {'PartNumber': 1, 'ETag': '"abc123"'}, {'PartNumber': 2, 'ETag': '"def456"'} ] ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=parts ) ``` -------------------------------- ### Perform Multipart Upload with KS3 SDK Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/CLAUDE.md Handle large file uploads using multipart upload. Two methods are shown: using the MultiPartUpload object directly or using Bucket API methods, which is recommended. This includes initiating, uploading parts, listing parts, and completing the upload. ```python # 方式一:使用 MultiPartUpload 对象 mp = bucket.initiate_multipart_upload('object_name') # 上传分段 mp.upload_part_from_file(file_part, part_num=1) # 完成上传 mp.complete_upload() # 方式二:直接使用 Bucket 方法(推荐) # 初始化分段上传 mp = bucket.initiate_multipart_upload('object_name') upload_id = mp.id # 上传分段 bucket.upload_part_from_file( key_name='object_name', upload_id=upload_id, fp=file_part, part_num=1 ) # 列出已上传的分片 result = bucket.list_parts( bucket_name=bucket.name, key_name='object_name', upload_id=upload_id ) # 完成上传 - 新方式:直接传入 Part 列表 ret = bucket.complete_multipart_upload( key_name='object_name', upload_id=upload_id, parts=result # 可以是 IterableResponseResult、Part 列表或字典列表 ) # 完成上传 - 传统方式:手动构建 XML xml_body = '\n' for part in result.data: xml_body += ' \n' xml_body += ' %d\n' % part.part_number xml_body += ' %s\n' % part.etag xml_body += ' \n' xml_body += '' ret = bucket.complete_multipart_upload( key_name='object_name', upload_id=upload_id, xml_body=xml_body ) ``` -------------------------------- ### Complete Multipart Upload - Using parts parameter (IterableResponseResult) Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md This recommended method uses the `parts` parameter, directly passing the result from `list_parts` (an IterableResponseResult object) to complete the multipart upload. ```APIDOC ## Complete Multipart Upload - Using parts parameter (IterableResponseResult) ### Description This recommended method uses the `parts` parameter, directly passing the result from `list_parts` (an IterableResponseResult object) to complete the multipart upload. This approach is concise and less prone to errors. ### Method `complete_multipart_upload` ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body - **key_name** (str) - Required - The object key name. - **upload_id** (str) - Required - The Upload ID for the multipart upload. - **parts** (IterableResponseResult) - Required - The result object from a `list_parts` call, containing information about uploaded parts. - **headers** (dict) - Optional - Custom request headers. ### Request Example ```python # List uploaded parts result = b.list_parts( bucket_name=b.name, key_name='object-key', upload_id='upload-id', part_number_marker=0 ) # Directly pass the list_parts result ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=result # Directly pass the IterableResponseResult ) ``` ### Response #### Success Response (200) Details of the completed multipart upload. #### Response Example ```json { "ETag": "\"some-etag\"", "Location": "http://bucket.ks3.com/object-key" } ``` ``` -------------------------------- ### Generate Upload URL Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Generates a pre-signed URL for uploading a file to a bucket. The URL has a specified expiration time. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket(bucket_name) # 新建对象key k = b.new_key(key_name) if k: url = k.get_presigned_url(60) # 60s 后该链接过期 print(url) ``` ```python k.get_presigned_url(1492073594, expires_in_absolute=True) # 1492073594为Unix Time ``` -------------------------------- ### Complete Multipart Upload with IterableResponseResult Parts Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md This is the recommended method for completing multipart uploads. It directly uses the result from list_parts, which is an IterableResponseResult, simplifying the process and reducing potential errors. ```python # 列出已上传的分片 result = b.list_parts( bucket_name=b.name, key_name='object-key', upload_id='upload-id', part_number_marker=0 ) # 直接传入 list_parts 返回的结果 ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=result # 直接传入 IterableResponseResult ) ``` -------------------------------- ### Complete Multipart Upload with Part Object List Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md Use this method when you have a list of Part objects, typically obtained from the 'data' attribute of a list_parts result. This approach is also part of the recommended 'parts' parameter usage. ```python from ks3.multipart import Part # 从 list_parts 获取 Part 对象列表 result = b.list_parts( bucket_name=b.name, key_name='object-key', upload_id='upload-id' ) # 使用 Part 对象列表 ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', parts=result.data # result.data 是 Part 对象列表 ) ``` -------------------------------- ### Copy Object Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Copies a file from a source bucket to a destination bucket. Requires read permissions on the source object. ```APIDOC ## Copy Object ### Description Copies a file from a source bucket to a destination bucket. Requires read permissions on the source object. ### Method ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) try: b = c.get_bucket('') b.copy_key('', '', '') except: pass ``` ``` -------------------------------- ### Upload Object Data Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Uploads a file from a specified directory to KS3, with options for setting file ACLs. It also covers uploading strings as values and using low-frequency storage. ```APIDOC ## Upload Object Data ### Description Uploads a file from a specified directory to KS3, with options for setting file ACLs. It also covers uploading strings as values and using low-frequency storage. ### Method ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) try: b = c.get_bucket('') k = b.new_key('') # object policy : 'private' or 'public-read' ret=k.set_contents_from_filename("", policy="private") if ret and ret.status == 200: print("上传成功") except: pass #异常处理 ``` ### Upload String as Value ```python k.set_contents_from_string('') ``` ### Upload with Low-Frequency Storage ```python # x-kss-storage-class有效值为"STANDARD"、"STANDARD_IA"。"STANDARD"表示标准存储,"STANDARD_IA"表示低频存储,如果不指定,默认为标准存储。 headers = {"x-kss-storage-class": "STANDARD_IA"} resp = k.set_contents_from_filename("", policy="private", headers=headers) ``` ### Upload Callback Requires setting `x-kss-callbackurl` and `x-kss-callbackbody` request headers. ```python headers = {"x-kss-callbackurl": "Callback_URL", "x-kss-callbackbody":"objectKey=${key}&etag=${etag}&uid=123"} ret = k.set_contents_from_filename("", headers=headers) ``` ### Fetch Network Resource Upload `x-kss-acl` can be set to `public-read` as needed; if not set, it defaults to private. ```python bucket = conn.get_bucket('') bucket.fetch_object('www-logo', source_url='http://fe.ksyun.com/project/resource/img/www-logo.png', headers={'x-kss-acl': 'public-read'}) ``` ``` -------------------------------- ### Complete Multipart Upload with XML Body Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/complete_multipart_usage.md Use this method when you need to manually construct the XML string for completing a multipart upload. Ensure the XML is correctly formatted according to the KS3 API. ```python from ks3.connection import Connection conn = Connection('', '', host='ks3-cn-beijing.ksyuncs.com') b = conn.get_bucket('') # 手动构建 XML xml_body = '\n' for part in result.data: xml_body += ' \n' xml_body += ' %d\n' % part.part_number xml_body += ' %s\n' % part.etag xml_body += ' \n' xml_body += '' # 完成分片上传 ret = b.complete_multipart_upload( key_name='object-key', upload_id='upload-id', xml_body=xml_body ) ``` -------------------------------- ### Copy Object Between Buckets Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Copy an object from a source bucket to a destination bucket. Ensure you have read permissions on the source object. ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) try: b = c.get_bucket('') b.copy_key('', '', '') except: pass ``` -------------------------------- ### Upload String Content to KS3 Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/GUIDE.md Upload a string directly as the content of a KS3 object. This is useful for small text-based data. ```python k.set_contents_from_string('') ``` -------------------------------- ### Download Object Data Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Downloads object data either as a string or to a file. ```APIDOC ## Download Object Data ### Description Downloads object data and returns it either as a string or saves it to a file. ### Method GET (Implicit) ### Endpoint `//` ### Code Example (Download as String) ```python from ks3.connection import Connection c = Connection('', '', host='', is_secure=False, domain_mode=False) b = c.get_bucket('') try: k = b.get_key('') s = k.get_contents_as_string().decode() print(s) except: pass # Exception handling ``` ### Code Example (Download to File) ```python # Save to file k.get_contents_to_filename("SAVED_FILE_PATH") ``` ``` -------------------------------- ### Upload Object Data Source: https://github.com/ks3sdk/ks3-python-sdk/blob/master/docs/README.md Uploads object data to a bucket. ```APIDOC ## Upload Object Data ### Description Uploads object data to a specified bucket. ### Method PUT (Implicit) ### Endpoint `//` ### Features - Upload Callback - Fetch Network Resource Upload ### Code Example (Basic Upload - Implicit) ```python # Assuming 'k' is a Key object obtained from a Bucket # k.set_contents_from_string('your object data') # or # k.set_contents_from_filename('local_file_path') ``` ```