獲取對象列表

功能說明

list_objects操作用于列出存儲桶中的全部對象，該操作返回最多1000個對象信息，可以通過設置過濾條件來列出存儲桶中符合特定條件的對象信息。

代碼示例

def list_objects(self):
    print('list_objects')
    response = self.s3_client.list_objects(
        Bucket='<your-bucket-name>',
        MaxKeys=50,  # list up to 50 key at a time
    )
    for obj in response['Contents']:
        print(obj["Key"])

如果 list 大于1000，則返回的結果中 isTruncated 為true，通過Marker參數可以指定下次讀取的起點。列舉所有對象的示例代碼如下：

def list_objects2(self):
    print('list_objects')
    objects = []
    response = self.s3_client.list_objects(
        Bucket=self.bucket,
        MaxKeys=100,
    )
    objects.extend(response['Contents'])
    while response['IsTruncated']:
        response = self.s3_client.list_objects(
            Bucket=self.bucket,
            MaxKeys=100,
            Marker=response['Contents'][-1]['Key']
        )
        objects.extend(response['Contents'])
?
    for obj in objects:
        print(obj["Key"])

請求參數

返回結果

上傳對象

功能說明

您可以使用put_object接口直接上傳文件。

代碼示例

def put_object(self):
    print('put_object')
    key = '<your-object-key>'
    local_path = '<file-path>'
    with open(local_path, 'rb') as f:
        resp = self.s3_client.put_object(
            Bucket='<your-bucket-name>',
            Key=key,
            Body=f,
            # ACL='public-read',
            # ContentType='text/json',
        )
        print(resp)

請求參數

返回結果

注意：put_object對文件大小有限制，最大能上傳5GB大小的文件，超過5GB需要使用分片上傳。

下載對象

功能說明

您可以使用get_object下載對象。

代碼示例

def get_object(self):
    print('get_object')
    key = '<your-object-key>'
    local_path = 'E:/ExampleObject.txt'
    resp = self.s3_client.get_object(
        Bucket='<your-bucket-name>',
        Key=key
    )
    body = resp["Body"]
    with open(local_path, 'wb') as f:
        for chunk in body:
            f.write(chunk)

請求參數

返回結果

復制對象

功能說明

您可以使用copy_object復制一個已經在媒體存儲中的對象。使用copy_object可以復制單個最大為5GB的對象。執行copy_object操作，必須具有對被拷貝對象的READ權限和對目標bucket的WRITE權限。

代碼示例

def copy_object(self):
    key = '<dst-object-key>'
    resp = self.s3_client.copy_object(
        Bucket='<dst-bucket-name>',
        Key=key,
        CopySource={'Bucket': '<source-bucket-name>', 'Key': '<source-object-key>'},
        ContentType="text/json",
        MetadataDirective='REPLACE',
    )
    print(resp)

文件比較大（超過1GB）的情況下，直接使用copyObject 可能會出現超時，需要使用分片復制的方式進行文件復制。TransferManager封裝了分片復制的接口，可以用于復制文件。

class TransferDemo(object):
    def __init__(self):
        config = botocore.config.Config(
            signature_version='s3v4',  # s3 or s3v4
        )
?
        session = botocore.session.get_session()
        self.s3_client = session.create_client(
            's3',
            aws_access_key_id='<your-access-key>',
            aws_secret_access_key='<your-secret-key>',
            endpoint_url='<your-endpoint>',
            config=config)
?
        MB = 1024 * 1024
        transConfig = s3transfer.manager.TransferConfig(
            multipart_threshold=5 * MB,  # 大于該值使用分片上傳
            multipart_chunksize=5 * MB,  # 分片大小
            max_request_concurrency=2,
            )
?
        # 設置帶寬，不填表示不限制
        # transConfig.max_bandwidth = 1 * MB
?
        self.transfer = s3transfer.manager.TransferManager(self.s3_client, transConfig)
?
    def copy(self):
        print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), "copy start")
        source={'Bucket': '<source-bucket-name>', 'Key': '<source-object-key>'}
        dstBucket = '<dst-bucket-name>'
        dstKey = '<dst-object-key>'
        # 擴展配置
        extraArgs = {'ContentType': 'text/plain', 'ACL': 'public-read'}
        future = self.transfer.copy(source, dstBucket, dstKey, extra_args=extraArgs)
        future.result()
        print(datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f'), "copy success")

請求參數

關于CopySource：

返回結果

刪除對象

功能說明

您可以使用delete_object接口刪除對象。

代碼示例

def delete_object(self):
    key = '<your-object-key>'
    resp = self.s3_client.delete_object(
        Bucket='<your-bucket-name>',
        Key=key,
    )
    print(resp)

請求參數

返回結果

根據返回碼判斷是否刪除成功，204表示刪除成功

批量刪除對象

功能說明

您可以使用delete_objects接口批量刪除多個對象，可以減少發起多個請求去刪除大量對象的花銷。delete_objects操作發起一個包含了最多1000個key的刪除請求，媒體存儲服務會對相應的對象逐個進行刪除，并且將刪除成功或者失敗的結果通過response返回。如果請求刪除的對象不存在，會返回已刪除的結果。

delete_objects操作返回包含verbose 和quiet兩種response模式。verbose response是默認的返回模式，該模式的返回結果包含了每個key的刪除結果。quiet response返回模式返回的結果僅包含了刪除失敗的key，對于一個完全成功的刪除操作，該返回模式不在相應消息體中返回任何信息。

代碼示例

def delete_objects(self):
    resp = self.s3_client.delete_objects(
        Bucket='<your-bucket-name>',
        Delete={
            'Objects': [
                {
                    'Key': 'ExampleObject.txt',
                },
                {
                    'Key': 'ExampleObject1.txt',
                },
            ],
        },
    )
    print(resp)

請求參數

返回結果

獲取對象元數據

功能說明

您可以使用head_object接口獲取對象元數據信息

代碼示例

def head_object():
    resp = s3_client.head_object(
        Bucket='<your-bucket-name>',
        Key='<your-object-key>',
    )
    print(resp)

請求參數

返回結果

設置對象訪問權限

功能說明

媒體存儲支持一組預先定義的授權，稱為Canned ACL。每個Canned ACL都有一組預定義的被授權者和權限，下表列出了相關的預定義授權含義。

您可以通過put_object_acl接口為媒體存儲服務中的對象設置ACL。對一個對象執行該操作需要具有WRITE_ACP權限。

代碼示例

def put_object_acl(self):
    key = '<your-object-key>'
    resp = self.s3_client.put_object_acl(
        Bucket='<your-bucket-name>',
        Key=key,
        ACL='public-read',
    )
    print(resp)

請求參數

返回結果

根據返回碼判斷是否設置成功，200表示成功。

獲取對象訪問權限

功能說明

您可以使用 get_object_acl操作獲取對象的access control list（ACL）信息。

代碼示例

def get_object_acl(self):
    key = '<your-object-key>'
    resp = self.s3_client.get_object_acl(
        Bucket='<your-bucket-name>',
        Key=key,
    )
    for grant in resp['Grants']:
        print(grant)

請求參數

返回結果

獲取對象標簽

功能說明

您可以使用get_object_tagging接口獲取對象標簽。

代碼示例

def get_object_tagging():
    resp = s3_client.get_object_tagging(
        Bucket='<your-bucket-name>',
        Key='<your-object-key>'
    )
    print(resp)

請求參數

返回結果

刪除對象標簽

功能說明

您可以使用delete_object_tagging接口刪除對象標簽。

代碼示例

def delete_object_tagging():
    resp = s3_client.delete_object_tagging(
        Bucket='<your-bucket-name>',
        Key='<your-object-key>'
    )
    print(resp)

請求參數

返回結果

根據返回碼判斷是否刪除成功，204表示刪除成功

設置對象標簽

功能說明

您可以使用put_object_tagging接口為對象設置標簽。標簽是一個鍵值對，每個對象最多可以有10個標簽。bucket的擁有者默認擁有給bucket中的對象設置標簽的權限，并且可以將權限授予其他用戶。每次執行PutObjectTagging操作會覆蓋對象已有的標簽信息。每個對象最多可以設置10個標簽，標簽Key和Value區分大小寫，并且Key不可重復。每個標簽的Key長度不超過128字節，Value長度不超過256字節。SDK通過HTTP header的方式設置標簽且標簽中包含任意字符時，需要對標簽的Key和Value做URL編碼。設置對象標簽信息不會更新對象的最新更改時間。

代碼示例

def put_object_tagging():
    resp = s3_client.put_object_tagging(
        Bucket='<your-bucket-name>',
        Key='<your-object-key>',
        Tagging={
            'TagSet': [
                {
                    'Key': 'key1',
                    'Value': 'value1'
                },
            ]
        }
    )
    print(resp)

請求參數

返回結果

根據返回碼判斷是否設置成功，200表示設置成功

生成預簽名URL

功能說明

您可以通過generate_presigned_url接口為一個指定對象生成一個預簽名的鏈接，用于下載或上傳對象。

代碼示例

• 生成一個預簽名的下載鏈接：

def generate_getobject_presigned_url(self):
    print('generate_getobject_presigned_url')
    key = '<your-object-key>'
    url = self.s3_client.generate_presigned_url(
        ClientMethod='get_object',
        Params={'Bucket': '<your-bucket-name>', 'Key': key},
        ExpiresIn=900)
    print(url)

• 訪問該預簽名下載鏈接可以直接下載該對象：

def getObjUsingPresignedUrl(self, presigned_url, download_path):
    import requests
    response = requests.get(presigned_url)
?
    if response.status_code == 200:
        with open(download_path, 'wb') as file:
            file.write(response.content)
        print(f"Downloaded object saved to: {download_path}")
    else:
        print(f"Failed to download object. Status code: {response.status_code}")
    
    return response.status_code, response.text

• 生成一個預簽名的上傳鏈接：

def generate_putobject_presigned_url(self):
    print('generate_putobject_presigned_url')
    key = '<your-object-key>'
    url = self.s3_client.generate_presigned_url(
        ClientMethod='put_object',
        Params={'Bucket': '<your-bucket-name>', 'Key': key},
        ExpiresIn=900)
?
    print(url)

• 通過該預簽名上傳鏈接可以直接上傳該對象：

def putObjUsingPresignedUrl(self, presigned_url, file_path):
    import requests
    with open(file_path, 'rb') as file:
        response = requests.put(presigned_url, data=file)
    
    print(f"Upload Response Status: {response.status_code}")
    return response.status_code, response.text

請求參數

返回結果

Post上傳

功能說明

generate_presigned_post接口為一個指定對象生成一個支持post方式上傳文件的參數集合，可以在前端使用post form-data的方式上傳文件。

代碼示例

def generate_postobject_presigned(self):
    print('generate_postobject_presigned')
    key = '<your-object-key>'
    Conditions = [
        ['starts-with', '$key', key],
    ]
?
    response = self.s3_client.generate_presigned_post(
        '<your-bucket-name>', 
        key,
        Fields={},
        Conditions=Conditions,
        ExpiresIn=3600)
?
    print(response)

請求參數

返回結果

前端使用方式如下：

<form action="<data.url>" method="POST" enctype="multipart/form-data">
    <input type="hidden" name="Policy" value="<data.fields['Policy']>" />
    <input type="hidden" name="X-Amz-Algorithm" value="<data.fields['X-Amz-Algorithm']>" />
    <input type="hidden" name="X-Amz-Credential" value="<data.fields['X-Amz-Credential']>" />
    <input type="hidden" name="X-Amz-Date" value="<data.fields['X-Amz-Date']>" />
    <input type="hidden" name="X-Amz-Signature" value="<data.fields['X-Amz-Signature']>" />
    <input type="hidden" name="bucket" value="<data.fields['bucket']>" />
    <input type="hidden" name="key" value="<data.fields['key']>" />
?
    <input type="file" name="file" value="" />
    <input type="submit" value="Submit" />
</form>

獲取多版本對象列表

功能說明

如果桶開啟了版本控制，您可以使用 list_object_versions接口列舉對象的版本，每次list操作最多返回1000個對象。

代碼示例

def list_object_versions(self):
    print('list_object_versions')
    response = self.s3_client.list_object_versions(
        Bucket='<your-bucket-name>',
        MaxKeys=50,  # list up to 50 key at a time
    )
    for obj in response['Versions']:
        print(obj["Key"])

請求參數

返回結果