收集必要資訊

下列變數會出現在範例中：

• bucket_name 必須是唯一且 DNS 安全的字串。 因為儲存區名稱在整個系統中是唯一的，因此如果此範例執行多次，將需要變更這些值。 請注意，名稱在刪除後會保留 10 到 15 分鐘。
• ibm_api_key_id 是服務認證中作為 apikey 找到的值。
• ibm_service_instance_id 是服務認證中作為 resource_instance_id 找到的值。
• endpoint_url 是服務端點 URL，包含 https:// 通訊協定。 此值並非 服務憑證 中的 endpoints 值。 如需端點的相關資訊，請參閱端點及儲存空間位置。
• LocationConstraint 是與 endpoint 值對應的 有效佈建代碼。

起始設定配置

此範例會建立 resource 物件。 資源提供 COS 的物件導向介面。 這容許比用戶端物件所提供的低階呼叫更高的抽象層次。

請注意，某些操作（例如 Aspera 高速傳輸）需要一個 client 物件。 Aspera 其本身需要 Python 版本 3.6。

舊版通知：對 Aspera 的支援已視為舊版功能。 相反地，請使用 Aspera Transfer SDK.

import ibm_boto3
from ibm_botocore.client import Config, ClientError

# Constants for IBM COS values
COS_ENDPOINT = "<endpoint>" # Current list avaiable at https://control.cloud-object-storage.cloud.ibm.com/v2/endpoints
COS_API_KEY_ID = "<api-key>" # eg "W00YixxxxxxxxxxMB-odB-2ySfTrFBIQQWanc--P3byk"
COS_INSTANCE_CRN = "<service-instance-id>" # eg "crn:v1:bluemix:public:cloud-object-storage:global:a/3bf0d9003xxxxxxxxxx1c3e97696b71c:d6f04d83-6c4f-4a62-a165-696756d63903::"

# Create resource
cos_resource = ibm_boto3.resource("s3",
    ibm_api_key_id=COS_API_KEY_ID,
    ibm_service_instance_id=COS_INSTANCE_CRN,
    config=Config(signature_version="oauth"),
    endpoint_url=COS_ENDPOINT
)

用戶端提供 COS S3 API 的低階介面。 這允許直接處理 HTTP 回應，而不是利用資源提供的抽象方法和屬性來存取標頭或 XML 回應有效負載所包含的資訊。

import ibm_boto3
from ibm_botocore.client import Config, ClientError

# Constants for IBM COS values
COS_ENDPOINT = "<endpoint>" # Current list avaiable at https://control.cloud-object-storage.cloud.ibm.com/v2/endpoints
COS_API_KEY_ID = "<api-key>" # eg "W00YixxxxxxxxxxMB-odB-2ySfTrFBIQQWanc--P3byk"
COS_INSTANCE_CRN = "<service-instance-id>" # eg "crn:v1:bluemix:public:cloud-object-storage:global:a/3bf0d9003xxxxxxxxxx1c3e97696b71c:d6f04d83-6c4f-4a62-a165-696756d63903::"

# Create client
cos_client = ibm_boto3.client("s3",
    ibm_api_key_id=COS_API_KEY_ID,
    ibm_service_instance_id=COS_INSTANCE_CRN,
    config=Config(signature_version="oauth"),
    endpoint_url=COS_ENDPOINT
)

建立新的儲存區

下面的範例使用低層次介面的用戶端。

LocationConstraint 的有效佈建碼清單可以在儲存空間類別手冊中參閱。

def create_bucket(bucket_name):
    print("Creating new bucket: {0}".format(bucket_name))
    try:
        cos_client.create_bucket(
            Bucket=bucket_name,
            CreateBucketConfiguration={
                "LocationConstraint":COS_BUCKET_LOCATION
            }
        )
        print("Bucket: {0} created!".format(bucket_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to create bucket: {0}".format(e))

建立新的文字檔

def create_text_file(bucket_name, item_name, file_text):
    print("Creating new item: {0}".format(item_name))
    try:
        cos_client.put_object(
            Bucket=bucket_name,
            Key=item_name,
            Body=file_text
        )
        print("Item: {0} created!".format(item_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to create text file: {0}".format(e))

列出可用的儲存區

def get_buckets():
    print("Retrieving list of buckets")
    try:
        buckets = cos_client.list_buckets()
        for bucket in buckets["Buckets"]:
            print("Bucket Name: {0}".format(bucket["Name"]))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to retrieve list buckets: {0}".format(e))

列出儲存區中的項目

def get_bucket_contents(bucket_name):
    print("Retrieving bucket contents from: {0}".format(bucket_name))
    try:
        files = cos_client.list_objects(Bucket=bucket_name)
        for file in files.get("Contents", []):
            print("Item: {0} ({1} bytes).".format(file["Key"], file["Size"]))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to retrieve bucket contents: {0}".format(e))

取得特定項目的檔案內容

def get_item(bucket_name, item_name):
    print("Retrieving item from bucket: {0}, key: {1}".format(bucket_name, item_name))
    try:
        file = cos_client.get_object(Bucket=bucket_name, Key=item_name)
        print("File Contents: {0}".format(file["Body"].read()))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to retrieve file contents: {0}".format(e))

從儲存區刪除項目

def delete_item(bucket_name, object_name):
    try:
        cos_client.delete_object(Bucket=bucket_name, Key=object_name)
        print("Item: {0} deleted!\n".format(object_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to delete object: {0}".format(e))

從儲存區刪除多個項目

刪除要求最多可以包含您要刪除的 1000 個金鑰。 雖然這對於降低每次要求的效能衝擊很有用，但在刪除許多鍵時要小心。 此外，也請考量物件的大小，以確保適當的效能。

def delete_items(bucket_name):
    try:
        delete_request = {
            "Objects": [
                { "Key": "deletetest/testfile1.txt" },
                { "Key": "deletetest/testfile2.txt" },
                { "Key": "deletetest/testfile3.txt" },
                { "Key": "deletetest/testfile4.txt" },
                { "Key": "deletetest/testfile5.txt" }
            ]
        }

        response = cos_client.delete_objects(
            Bucket=bucket_name,
            Delete=delete_request
        )

        print("Deleted items for {0}\n".format(bucket_name))
        print(json.dumps(response.get("Deleted"), indent=4))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to copy item: {0}".format(e))

刪除儲存區

def delete_bucket(bucket_name):
    print("Deleting bucket: {0}".format(bucket_name))
    try:
        cos_client.delete_bucket(Bucket=bucket_name)
        print("Bucket: {0} deleted!".format(bucket_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to delete bucket: {0}".format(e))

執行多部分上傳

def multi_part_upload(bucket_name, item_name, file_path):
    try:
        print("Starting file transfer for {0} to bucket: {1}\n".format(item_name, bucket_name))
        # set 5 MB chunks
        part_size = 1024 * 1024 * 5

        # set threadhold to 15 MB
        file_threshold = 1024 * 1024 * 15

        # set the transfer threshold and chunk size
        transfer_config = ibm_boto3.s3.transfer.TransferConfig(
            multipart_threshold=file_threshold,
            multipart_chunksize=part_size
        )

        # the upload_fileobj method will automatically execute a multi-part upload
        # in 5 MB chunks for all files over 15 MB
        with open(file_path, "rb") as file_data:
            cos_client.upload_fileobj(
                Bucket=bucket_name,
                Key=item_name,
                Fileobj=file_data,
                Config=transfer_config
            )

        print("Transfer for {0} Complete!\n".format(item_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to complete multi-part upload: {0}".format(e))

def multi_part_upload_manual(bucket_name, item_name, file_path):
    try:
        # create client object
        cos_client = ibm_boto3.client("s3",
            ibm_api_key_id=COS_API_KEY_ID,
            ibm_service_instance_id=COS_SERVICE_CRN,
            config=Config(signature_version="oauth"),
            endpoint_url=COS_ENDPOINT
        )

        print("Starting multi-part upload for {0} to bucket: {1}\n".format(item_name, bucket_name))

        # initiate the multi-part upload
        mp = cos_client.create_multipart_upload(
            Bucket=bucket_name,
            Key=item_name
        )

        upload_id = mp["UploadId"]

        # min 20MB part size
        part_size = 1024 * 1024 * 20
        file_size = os.stat(file_path).st_size
        part_count = int(math.ceil(file_size / float(part_size)))
        data_packs = []
        position = 0
        part_num = 0

        # begin uploading the parts
        with open(file_path, "rb") as file:
            for i in range(part_count):
                part_num = i + 1
                part_size = min(part_size, (file_size - position))

                print("Uploading to {0} (part {1} of {2})".format(item_name, part_num, part_count))

                file_data = file.read(part_size)

                mp_part = cos_client.upload_part(
                    Bucket=bucket_name,
                    Key=item_name,
                    PartNumber=part_num,
                    Body=file_data,
                    ContentLength=part_size,
                    UploadId=upload_id
                )

                data_packs.append({
                    "ETag":mp_part["ETag"],
                    "PartNumber":part_num
                })

                position += part_size

        # complete upload
        cos_client.complete_multipart_upload(
            Bucket=bucket_name,
            Key=item_name,
            UploadId=upload_id,
            MultipartUpload={
                "Parts": data_packs
            }
        )
        print("Upload for {0} Complete!\n".format(item_name))
    except ClientError as be:
        # abort the upload
        cos_client.abort_multipart_upload(
            Bucket=bucket_name,
            Key=item_name,
            UploadId=upload_id
        )
        print("Multi-part upload aborted for {0}\n".format(item_name))
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to complete multi-part upload: {0}".format(e))

使用 TransferManager 的大型物件上傳

TransferManager 提供另一種方式來執行大型檔案傳送，方法是在必須設定配置參數時自動併入多部分上傳。

def upload_large_file(bucket_name, item_name, file_path):
    print("Starting large file upload for {0} to bucket: {1}".format(item_name, bucket_name))

    # set the chunk size to 5 MB
    part_size = 1024 * 1024 * 5

    # set threadhold to 5 MB
    file_threshold = 1024 * 1024 * 5

    # Create client connection
    cos_client = ibm_boto3.client("s3",
        ibm_api_key_id=COS_API_KEY_ID,
        ibm_service_instance_id=COS_SERVICE_CRN,
        config=Config(signature_version="oauth"),
        endpoint_url=COS_ENDPOINT
    )

    # set the transfer threshold and chunk size in config settings
    transfer_config = ibm_boto3.s3.transfer.TransferConfig(
        multipart_threshold=file_threshold,
        multipart_chunksize=part_size
    )

    # create transfer manager
    transfer_mgr = ibm_boto3.s3.transfer.TransferManager(cos_client, config=transfer_config)

    try:
        # initiate file upload
        future = transfer_mgr.upload(file_path, bucket_name, item_name)

        # wait for upload to complete
        future.result()

        print ("Large file upload complete!")
    except Exception as e:
        print("Unable to complete large file upload: {0}".format(e))
    finally:
        transfer_mgr.shutdown()

列出儲存區中的項目（第 2 版）

S3.Client 物件具有已更新的方法來列出內容 (list_objects_v2)。 此方法可讓您限制傳回的記錄數，並分批次擷取記錄。 這可能有助於在應用程式內將您的結果進行分頁，並改善效能。

def get_bucket_contents_v2(bucket_name, max_keys):
    print("Retrieving bucket contents from: {0}".format(bucket_name))
    try:
        # create client object
        cos_client = ibm_boto3.client("s3",
            ibm_api_key_id=COS_API_KEY_ID,
            ibm_service_instance_id=COS_SERVICE_CRN,
            config=Config(signature_version="oauth"),
            endpoint_url=COS_ENDPOINT)

        more_results = True
        next_token = ""

        while (more_results):
            response = cos_client.list_objects_v2(Bucket=bucket_name, MaxKeys=max_keys, ContinuationToken=next_token)
            files = response["Contents"]
            for file in files:
                print("Item: {0} ({1} bytes).".format(file["Key"], file["Size"]))

            if (response["IsTruncated"]):
                next_token = response["NextContinuationToken"]
                print("...More results in next batch!\n")
            else:
                more_results = False
                next_token = ""

    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to retrieve bucket contents: {0}".format(e))

建立備份政策

# Config values
api_key = "<API_KEY>"
vault_crn = "<SERVICE_INSTANCE_ID>"
source_bucket_name = "<BACKUP_VAULT_NAME>"
policy_name = "<POLICY_NAME>"

# Authenticator and client setup
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Create policy
create_backup_policy = rc_client.create_backup_policy(
        bucket=source_bucket_name,
        policy_name=policy_name,
        target_backup_vault_crn=vault_crn,
        backup_type="continuous",
        initial_retention={"delete_after_days": 1}
    )

# Print response
print(f" Policy created: { create_backup_policy }")

列出備份政策

# Config values
api_key = "<API_KEY>"
source_bucket_name = "<BACKUP_VAULT_NAME>"

# Authenticator and client setup
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# List all backup policies
list_response = rc_client.list_backup_policies(bucket=source_bucket_name)

print("\nList of backup policies:")
for policy in list_response.result.get("backup_policies", []):
    print(policy)

取得備份政策

# Config
api_key = "<API_KEY>"
source_bucket_name = "<SOURCE_BUCKET_NAME>"
backup_vault_crn = "<BACKUP_VAULT_CRN>"
policy_name = "<POLICY_NAME>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Create backup policy
create_backup_policy_response = rc_client.create_backup_policy(
    bucket=source_bucket_name,
    policy_name=policy_name,
    target_backup_vault_crn=backup_vault_crn,
    backup_type="continuous",
    initial_retention={"delete_after_days": 1}
)

# Extract policy ID
policy_id = create_backup_policy_response.result.get("policy_id")

get_backup_policy_response = rc_client.get_backup_policy(
    bucket=source_bucket_name,
    policy_id=policy_id
)

print("\nFetched Backup Policy Details:")
print(get_backup_policy_response.result)

刪除備份原則

# Config
api_key = "<API_KEY>"
source_bucket_name = "<SOURCE_BUCKET_NAME>"
policy_id = "<POLICY_ID>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Delete the backup policy
delete_backup_policy_response = rc_client.delete_backup_policy(
    bucket=source_bucket_name,
    policy_id=policy_id
)

print(f"Backup policy '{policy_id}' deleted successfully.")

建立備份儲存庫

# Config
api_key = "<API_KEY>"
service_instance_id = "<SERVICE_INSTANCE_ID>"
backup_vault_name = "<BACKUP_VAULT_NAME>"
region = "<REGION>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Create a backup vault
create_backup_vault_response = rc_client.create_backup_vault(
    service_instance_id=service_instance_id,
    backup_vault_name=backup_vault_name,
    region=region
)

# Output result
print("Backup vault created:")
print(create_backup_vault_response.result)

列出備份儲存庫

# Config
api_key = "<API_KEY>"
service_instance_id = "<SERVICE_INSTANCE_ID>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# List backup vaults
list_backup_vaults_response = rc_client.list_backup_vaults(
    service_instance_id=service_instance_id
)

print("List of backup vaults:")
print(list_backup_vaults_response.result)

取得備份儲存庫

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Get backup vault
get_backup_vault = rc_client.get_backup_vault(
    backup_vault_name=backup_vault_name
)

# Output result
print("Backup vault details:")
print(get_backup_vault.result)

更新備份儲存庫

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Update backup vault settings (disable activity tracking and metrics monitoring)
backup_vault_patch = {
    "activity_tracking": {"management_events": True},
    "metrics_monitoring": {"usage_metrics_enabled": True},
}

update_backup_vault_response = rc_client.update_backup_vault(
    backup_vault_name=backup_vault_name,
    backup_vault_patch=backup_vault_patch
)

# Output result
print("Backup vault updated successfully.")
print(update_backup_vault_response)

刪除備份儲存庫

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Delete the backup vault
delete_vault_response = rc_client.delete_backup_vault(
    backup_vault_name=backup_vault_name
)

# Output result
print(f"Successfully deleted backup vault '{delete_vault_response}'.")

列名復原範圍

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# List recovery ranges
recovery_ranges_response = rc_client.list_recovery_ranges(
    backup_vault_name=backup_vault_name
)

# Output recovery range results
print("Recovery Ranges:")
print(recovery_ranges_response.result)

取得復原範圍

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"
recovery_range_id = "<RECOVERY_RANGE_ID>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

get_recovery_range_response = rc_client.get_source_resource_recovery_range(
    backup_vault_name=backup_vault_name,
    recovery_range_id=recovery_range_id
)
print("Recovery Range Details:")
print(get_recovery_range_response.result)

更新復原範圍

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"
recovery_range_id = "<RECOVERY_RANGE_ID>"

# Setup authenticator and client
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

recovery_range_patch_model = {}
recovery_range_patch_model['retention'] = {"delete_after_days": 99}

patch_response = rc_client.patch_source_resource_recovery_range(
    backup_vault_name=backup_vault_name,
    recovery_range_id=recovery_range_id,
    recovery_range_patch=recovery_range_patch_model
)
print("Patch Response Details:")
print(patch_response)

啟動還原

# Configuration
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"
target_bucket_crn = "<TARGET_BUCKET_CRN>"
recovery_range_id = "<RECOVERY_RANGE_ID>"
restore_point_in_time = "<RESTORE_POINT_IN_TIME>"

# Setup authenticator and clients
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Initiate restore
create_restore = rc_client.create_restore(
    backup_vault_name=backup_vault_name,
    recovery_range_id=recovery_range_id,
    restore_type="in_place",
    restore_point_in_time=restore_point_in_time,
    target_resource_crn=target_bucket_crn
)
print(f"Restore initiated : {create_restore}")

列名還原

# Config
api_key = "<API_KEY>"
backup_vault_name = "<BACKUP_VAULT_NAME>"

# Setup authenticator and clients
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# List restore operations
get_store = rc_client.get_restore(
    backup_vault_name=backup_vault_name)

print("Restore response:")
print(get_store.result)

取得還原詳細資訊

# Config
api_key = "<API_KEY>"
source_bucket_name = "<SOURCE_BUCKET_NAME>"
backup_vault_crn = "<BACKUP_VAULT_CRN>"
backup_vault_name = "<BACKUP_VAULT_NAME>"
target_bucket_crn = "<TARGET_BUCKET_CRN>"
recovery_range_id = "<RECOVERY_RANGE_ID>"
restore_point_in_time = "<RESTORE_POINT_IN_TIME>"

# Setup authenticator and clients
authenticator = IAMAuthenticator(apikey=api_key)
rc_client = ResourceConfigurationV1(authenticator=authenticator)

# Create restore
create_restore = rc_client.create_restore(
    backup_vault_name=backup_vault_name,
    recovery_range_id=recovery_range_id,
    restore_type="in_place",
    restore_point_in_time=restore_point_in_time,
    target_resource_crn=target_bucket_crn
)

restore_id = create_restore.result["restore_id"]

# List restore operations
get_store = rc_client.get_restore(
    backup_vault_name=backup_vault_name, restore_id=restore_id)

print("Restore response:")
print(get_store.result)

建立一個啟用物件鎖定功能的新 COS 儲存桶

 def create_bucket_with_objectlock(bucket_name):
         cos_cli.create_bucket(
             Bucket=bucket_name,
             ObjectLockEnabledForBucket=True,
         )
         print("Bucket: {0} created with objectlock enabled".format(bucket_name))

將物件鎖定設定與合規模式套用至 COS 儲存桶

def objectlock_configuration_on_bucket(bucket_name):

    # Putting default retenion on the COS bucket.
    default_retention_rule = {'DefaultRetention': {'Mode': 'COMPLIANCE', 'Years': 1}}
    object_lock_config = {'ObjectLockEnabled': 'Enabled', 'Rule': default_retention_rule}
    cos_cli.put_object_lock_configuration(Bucket=bucket_name, ObjectLockConfiguration=object_lock_config)

將物件鎖定設定與治理模式套用至 COS 儲存桶

def objectlock_configuration_with_governance_mode_on_bucket(bucket_name):

    # Putting default retenion on the COS bucket with governance mode.
    default_retention_rule = {'DefaultRetention': {'Mode': 'GOVERNANCE', 'Years': 1}}
    object_lock_config = {'ObjectLockEnabled': 'Enabled', 'Rule': default_retention_rule}
    cos_cli.put_object_lock_configuration(Bucket=bucket_name, ObjectLockConfiguration=object_lock_config)

取得 COS 儲存桶的物件鎖定設定

def objectlock_configuration_with_governance_mode_on_bucket(bucket_name):

    # Reading the objectlock configuration set on the bucket.
    response = cos_cli.get_object_lock_configuration(Bucket=bucket_name)
    print("Objectlock Configuration for {0} =>".format(bucket_name))
    print(response.ObjectLockConfiguration)

將具有治理模式的物件上傳至 COS 儲存桶

def upload_object_with_governance_mode(bucket_name,object_name,object_content):
        cos_cli.put_object(
            Bucket=bucket_name,
            Key=object_name,
            Body=object_content,
            ObjectLockMode='GOVERNANCE',
            ObjectLockRetainUntilDate=datetime(2025, 11, 15)
        )
        print("Object: {0} uploaded!".format(object_name))

在物件上啟用物件鎖保留與合規模式

def objectlock_retention(bucket_name,object_name):
        # Put objectlock retenion on the  object uploaded to the bucket.
        date = datetime.now()+timedelta(seconds=5)
        retention_rule = {'Mode': 'COMPLIANCE', 'RetainUntilDate': date}
        cos_cli.put_object_retention(Bucket=bucket_name, Key=object_name, Retention=retention_rule)

在物件上啟用物件鎖保留與治理模式

def objectlock_retention_with_governance_mode(bucket_name,object_name):
        # Put objectlock retenion with governance mode on the  object uploaded to the bucket.
        date = datetime.now()+timedelta(seconds=5)
        retention_rule = {'Mode': 'GOVERNANCE', 'RetainUntilDate': date}
        cos_cli.put_object_retention(Bucket=bucket_name, Key=object_name, Retention=retention_rule)

取得物件鎖保留

def objectlock_retention_with_governance_mode(bucket_name,object_name):
        # Get objectlock retention of the above object.
        response = cos_cli.get_object_retention(Bucket=bucket_name, Key=object_name)
        print("Objectlock Retention for {0}=>".format(object_name))
        print(response.Retention)

將物件鎖定為法律保留狀態

def objectlock_legal_hold(bucket_name,object_name):
        # Setting the objectlock legal-hold status to ON.
        cos_cli.put_object_legal_hold(Bucket=bucket_name, Key=object_name, legal_hold={'Status': 'ON'})

取得物件鎖定法律保留

def objectlock_legal_hold(bucket_name,object_name):
        # Get objectlock retention of the above object.
        response = cos_cli.get_object_legal_hold(Bucket=bucket_name, Key=object_name)
        print("Objectlock legal-hold for {0}=>".format(object_name))
        print(response.legal_hold)

使用繞過治理刪除具有物件鎖治理模式的物件

def delete_object_with_bypass_governance(bucket_name,object_name):
        # Deleting an object with retention using bypass governance
        cos_cli.delete_object(Bucket=bucket_name, Key=object_name, BypassGovernanceRetention=True)

開始之前

為了建立儲存區並啟用 Key Protect，需要下列項目：

• 已佈建 Key Protect 服務
• 根金鑰可用（已產生或已匯入）

擷取根金鑰 CRN

1. 擷取 Key Protect 服務的實例 ID
2. 使用 Key Protect API 來擷取所有可用金鑰
   • 您可以使用 curl 指令或 API REST 用戶端（例如 Postman）來存取 Key Protect API。
3. 擷取您用來在儲存區上啟用 Key Protect 的根金鑰 CRN。 CRN 看起來如下：

crn:v1:bluemix:public:kms:us-south:a/3d624cd74a0dea86ed8efe3101341742:90b6a1db-0fe1-4fe9-b91e-962c327df531:key:0bg3e33e-a866-50f2-b715-5cba2bc93234

建立儲存區並啟用 key-protect

COS_KP_ALGORITHM = "<algorithm>"
COS_KP_ROOTKEY_CRN = "<root-key-crn>"

# Create a new bucket with key protect (encryption)
def create_bucket_kp(bucket_name):
    print("Creating new encrypted bucket: {0}".format(bucket_name))
    try:
        cos_client.create_bucket(
            Bucket=bucket_name,
            CreateBucketConfiguration={
                "LocationConstraint":COS_BUCKET_LOCATION
            },
            IBMSSEKPEncryptionAlgorithm=COS_KP_ALGORITHM,
            IBMSSEKPCustomerRootKeyCrn=COS_KP_ROOTKEY_CRN
        )
        print("Encrypted Bucket: {0} created!".format(bucket_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to create encrypted bucket: {0}".format(e))

起始設定 AsperaTransferManager

在初始化 AsperaTransferManager 之前，請確認您有一個可運作的 client ( 而不是 resource 或 session) 物件。

import ibm_boto3
from ibm_botocore.client import Config
from ibm_s3transfer.aspera.manager import AsperaTransferManager

COS_ENDPOINT = "<endpoint>" # Current list avaiable at https://control.cloud-object-storage.cloud.ibm.com/v2/endpoints
COS_API_KEY_ID = "<api-key>"
COS_RESOURCE_CRN = "<resource-instance-id>"
COS_BUCKET_LOCATION = "<location>"

# Create resource
cos_client = ibm_boto3.client("s3",
    ibm_api_key_id=COS_API_KEY_ID,
    ibm_service_instance_id=COS_RESOURCE_CRN,
    config=Config(signature_version="oauth"),
    endpoint_url=COS_ENDPOINT
)

transfer_manager = AsperaTransferManager(cos)

您需要提供 IAM API 金鑰，以進行 Aspera 高速傳送。 目前 不 支援 HMAC 認證。 如需 IAM 的相關資訊，請按一下這裡。

為了得到最高的傳輸量，請將傳送分割成指定數目的平行階段作業 ，以傳送大小由 threshold 值定義的資料區塊。

使用多個階段作業的一般配置應該是：

• 2500 MBps 目標率
• 100 MB 臨界值（此為大部分應用程式的建議值）

ms_transfer_config = AsperaConfig(multi_session="all",
                                  target_rate_mbps=2500,
                                  multi_session_threshold_mb=100)

在上述範例中，SDK 會大量產生足夠的階段作業，以嘗試達到目標率 2500 MBps。

也可以在 SDK 中明確配置階段作業管理。 在想要更精確控制網路使用率的情況下，這十分有用。

使用明確的多個階段作業的一般配置應該是：

• 2 或 10 個階段作業
• 100 MB 臨界值（此為大部分應用程式的建議值）

from ibm_s3transfer.aspera.manager import AsperaConfig
# Configure 2 sessions for transfer
ms_transfer_config = AsperaConfig(multi_session=2,
                                  multi_session_threshold_mb=100)

# Create the Aspera Transfer Manager
transfer_manager = AsperaTransferManager(client=client,
                                         transfer_config=ms_transfer_config)

為了在大多數情況下獲得最佳效能，請務必使用多個階段，以盡量減少任何與實體化 Aspera 高速傳輸相關的處理。 如果您的網路容量至少為 1 Gbps，您應該使用 10 個階段作業。 較低頻寬的網路應該使用兩個階段作業。

檔案上傳

bucket_name = "<bucket-name>"
upload_filename = "<absolute-path-to-file>"
object_name = "<item-name>"

# Create Transfer manager
with AsperaTransferManager(client) as transfer_manager:

    # Perform upload
    future = transfer_manager.upload(upload_filename, bucket_name, object_name)

    # Wait for upload to complete
    future.result()

檔案下載

bucket_name = "<bucket-name>"
download_filename = "<absolute-path-to-file>"
object_name = "<object-to-download>"

# Create Transfer manager
with AsperaTransferManager(client) as transfer_manager:

    # Get object with Aspera
    future = transfer_manager.download(bucket_name, object_name, download_filename)

    # Wait for download to complete
    future.result()

目錄上傳

bucket_name = "<bucket-name>"
# THIS DIRECTORY MUST EXIST LOCALLY, and have objects in it.
local_upload_directory = "<absolute-path-to-directory>"
# THIS SHOULD NOT HAVE A LEADING "/"
remote_directory = "<object prefix>"

# Create Transfer manager
with AsperaTransferManager(client) as transfer_manager:

    # Perform upload
    future = transfer_manager.upload_directory(local_upload_directory, bucket_name, remote_directory)

    # Wait for upload to complete
    future.result()

目錄下載

bucket_name = "<bucket-name>"
# THIS DIRECTORY MUST EXIST LOCALLY
local_download_directory = "<absolute-path-to-directory>"
remote_directory = "<object prefix>"

# Create Transfer manager
with AsperaTransferManager(client) as transfer_manager:

    # Get object with Aspera
    future = transfer_manager.download_directory(bucket_name, remote_directory, local_download_directory)

    # Wait for download to complete
    future.result()

使用訂閱者

訂閱者藉由連接自訂回呼方法而提供對於傳送的觀察。 所有傳送會在下列階段之間轉移：

Queued - In Progress - Done

每個階段有三個可用的訂閱者：

• CallbackOnQueued() - 有新的傳送新增至 AsperaTransferManager 時呼叫
• CallbackOnProgress() - 傳送已傳輸資料時呼叫（傳送進行中時會反覆地發動）。
• CallbackOnDone() - 傳送完成之後呼叫

bucket_name = "<bucket-name>"
local_download_directory = "<absolute-path-to-directory>"
remote_directory = "<object prefix>"

# Subscriber callbacks
class CallbackOnQueued(AsperaBaseSubscriber):
    def __init__(self):
        pass

    def on_queued(self, future, **kwargs):
        print("Directory download queued.")

class CallbackOnProgress(AsperaBaseSubscriber):
    def __init__(self):
        pass

    def on_progress(self, future, bytes_transferred, **kwargs):
        print("Directory download in progress: %s bytes transferred" % bytes_transferred)

class CallbackOnDone(AsperaBaseSubscriber):
    def __init__(self):
        pass

    def on_done(self, future, **kwargs):
        print("Downloads complete!")

# Create Transfer manager
transfer_manager = AsperaTransferManager(client)

# Attach subscribers
subscribers = [CallbackOnQueued(), CallbackOnProgress(), CallbackOnDone()]

# Get object with Aspera
future = transfer_manager.download_directory(bucket_name, remote_directory, local_download_directory, None, subscribers)

# Wait for download to complete
future.result()

Directory download queued.
Directory download in progress: 5632 bytes transferred
Directory download in progress: 1047552 bytes transferred
...
Directory download in progress: 53295130 bytes transferred
Directory download in progress: 62106855 bytes transferred
Download complete!

暫停/繼續/取消

SDK 可讓您透過 AsperaTransferFuture 物件的下列方法來管理檔案/目錄傳送的進度：

• pause()
• resume()
• cancel()

呼叫上述任一種方法時，不會產生任何副作用。 SDK 會處理適當的清除及整理。

# Create Transfer manager
bucket_name = "<bucket-name>"
local_download_directory = "<absolute-path-to-directory>"
remote_directory = "<object prefix>"

with AsperaTransferManager(client) as transfer_manager:

    # download a directory with Aspera
    future = transfer_manager.download_directory(bucket_name, remote_directory, local_download_directory, None, None)

    # pause the transfer
    future.pause()

    # resume the transfer
    future.resume()

    # cancel the transfer
    future.cancel()

疑難排解 Aspera 問題

問題: 除了 3.6 之外，使用任何 Python 版本的開發人員在安裝或使用 Aspera SDK時可能會遇到失敗。

原因： 如果您的環境中安裝了不同版本的 Python，那麼您在嘗試安裝 Aspera SDK 時可能會遇到安裝失敗的情況。 這可能是遺漏 DLL 檔案或路徑中有錯誤的 DLL 所致。

**解決方案：**解決此問題的首要步驟是重新安裝 Aspera 程式庫。 安裝期間可能有失敗。 因此，這可能已影響 DLL 檔案。 如果那未解決問題，則您必須更新 Python 的版本。 如果您無法這麼做，那麼您可以使用安裝 Intel® Distribution for Python *。 這應該可讓您在 Python 3.6.x 上安裝 Aspera SDK，而不會有任何問題。

使用 PUT 更新 meta 資料

注意： PUT 請求會覆蓋物件的現有內容，因此必須先下載，然後再以新的元資料重新上傳。

def update_metadata_put(bucket_name, item_name, key, value):
    try:
        # retrieve the existing item to reload the contents
        response = cos_client.get_object(Bucket=bucket_name, Key=item_name)
        existing_body = response.get("Body").read()

        # set the new metadata
        new_metadata = {
            key: value
        }

        cos_client.put_object(Bucket=bucket_name, Key=item_name, Body=existing_body, Metadata=new_metadata)

        print("Metadata update (PUT) for {0} Complete!\n".format(item_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        log_error("Unable to update metadata: {0}".format(e))

使用 COPY 更新 meta 資料

def update_metadata_copy(bucket_name, item_name, key, value):
    try:
        # set the new metadata
        new_metadata = {
            key: value
        }

        # set the copy source to itself
        copy_source = {
            "Bucket": bucket_name,
            "Key": item_name
        }

        cos_client.copy_object(Bucket=bucket_name, Key=item_name, CopySource=copy_source, Metadata=new_metadata, MetadataDirective="REPLACE")

        print("Metadata update (COPY) for {0} Complete!\n".format(item_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        log_error("Unable to update metadata: {0}".format(e))

將保護配置新增至現有儲存區

在保護期間過期，並移除物件上的所有合法保留之前，無法刪除寫入受保護儲存區的物件。 除非在建立物件時提供物件特定值，否則會將儲存區的預設保留值提供給物件。 不再保留的受保護儲存區物件（保留期間已過期，而物件沒有任何合法保留），在被改寫時會再次保留。 新的保留期間可以提供為物件改寫要求的一部分，否則會將儲存區的預設保留時間提供給物件。

保留期間設定 MinimumRetention、DefaultRetention 及 MaximumRetention 的最小及最大支援值為 0 天及 365243 天 (1000 年)。

def add_protection_configuration_to_bucket(bucket_name):
    try:
        new_protection_config = {
            "Status": "Retention",
            "MinimumRetention": {"Days": 10},
            "DefaultRetention": {"Days": 100},
            "MaximumRetention": {"Days": 1000}
        }

        cos_client.put_bucket_protection_configuration(Bucket=bucket_name, ProtectionConfiguration=new_protection_config)

        print("Protection added to bucket {0}\n".format(bucket_name))
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to set bucket protection config: {0}".format(e))

檢查儲存區的保護

def get_protection_configuration_on_bucket(bucket_name):
    try:
        response = cos_client.get_bucket_protection_configuration(Bucket=bucket_name)
        protection_config = response.get("ProtectionConfiguration")

        print("Bucket protection config for {0}\n".format(bucket_name))
        print(protection_config)
        print("\n")
    except ClientError as be:
        print("CLIENT ERROR: {0}\n".format(be))
    except Exception as e:
        print("Unable to get bucket protection config: {0}".format(e))

上傳受保護物件

不再保留的受保護儲存區物件（保留期間已過期，而物件沒有任何合法保留），在被改寫時會再次保留。 新的保留期間可以提供為物件改寫要求的一部分，否則會將儲存區的預設保留時間提供給物件。

def put_object_add_legal_hold(bucket_name, object_name, file_text, legal_hold_id):
    print("Add legal hold {0} to {1} in bucket {2} with a putObject operation.\n".format(legal_hold_id, object_name, bucket_name))
    cos_client.put_object(
        Bucket=bucket_name,
        Key=object_name,
        Body=file_text,
        RetentionLegalHoldId=legal_hold_id)
    print("Legal hold {0} added to object {1} in bucket {2}\n".format(legal_hold_id, object_name, bucket_name))

def copy_protected_object(source_bucket_name, source_object_name, destination_bucket_name, new_object_name):
    print("Copy protected object {0} from bucket {1} to {2}/{3}.\n".format(source_object_name, source_bucket_name, destination_bucket_name, new_object_name))

    copy_source = {
        "Bucket": source_bucket_name,
        "Key": source_object_name
    }

    cos_client.copy_object(
        Bucket=destination_bucket_name,
        Key=new_object_name,
        CopySource=copy_source,
        RetentionDirective="Copy"
    )

    print("Protected object copied from {0}/{1} to {2}/{3}\n".format(source_bucket_name, source_object_name, destination_bucket_name, new_object_name));

def complete_multipart_upload_with_retention(bucket_name, object_name, upload_id, retention_period):
    print("Completing multi-part upload for object {0} in bucket {1}\n".format(object_name, bucket_name))
    cos_client.complete_multipart_upload(
        Bucket=bucket_name,
        Key=object_name,
        MultipartUpload={
            "Parts":[{
                "ETag": part["ETag"],
                "PartNumber": 1
            }]
        },
        UploadId=upload_id,
        RetentionPeriod=retention_period
    )

    print("Multi-part upload completed for object {0} in bucket {1}\n".format(object_name, bucket_name))

def upload_file_with_retention(bucket_name, object_name, path_to_file, retention_period):
    print("Uploading file {0} to object {1} in bucket {2}\n".format(path_to_file, object_name, bucket_name))

    args = {
        "RetentionPeriod": retention_period
    }

    cos_client.upload_file(
        Filename=path_to_file,
        Bucket=bucket_name,
        Key=object_name,
        ExtraArgs=args
    )

    print("File upload complete to object {0} in bucket {1}\n".format(object_name, bucket_name))

新增或移除受保護物件的合法保留

物件可支援 100 個合法保留：

• 合法保留 ID 是長度上限為 64 個字元的字串，且長度下限為 1 個字元。 有效字元為字母、數字及符號 !、_、.、*、(、) 及 -。
• 如果給定合法保留新增數超過物件上的 100 個合法保留總數，則不會新增合法保留，並會傳回 400 錯誤。
• 如果 ID 太長，則不會將它新增至物件，並會傳回 400 錯誤。
• 如果 ID 包含無效的字元，則不會將它新增至物件，且會傳回 400 錯誤。
• 如果已在物件上使用 ID，則不會修改現有的合法保留，而且回應會指出已在使用 ID，且發生 409 錯誤。
• 如果物件沒有保留期間 meta 資料，則會傳回 400 錯誤，且不容許新增或移除合法保留。

若要新增或移除合法保留，您必須具有此儲存區的 Manager 許可權。

def add_legal_hold_to_object(bucket_name, object_name, legal_hold_id):
    print("Adding legal hold {0} to object {1} in bucket {2}\n".format(legal_hold_id, object_name, bucket_name))

    cos_client.add_legal_hold(
        Bucket=bucket_name,
        Key=object_name,
        RetentionLegalHoldId=legal_hold_id
    )

    print("Legal hold {0} added to object {1} in bucket {2}!\n".format(legal_hold_id, object_name, bucket_name))

def delete_legal_hold_from_object(bucket_name, object_name, legal_hold_id):
    print("Deleting legal hold {0} from object {1} in bucket {2}\n".format(legal_hold_id, object_name, bucket_name))

    cos_client.delete_legal_hold(
        Bucket=bucket_name,
        Key=object_name,
        RetentionLegalHoldId=legal_hold_id
    )

    print("Legal hold {0} deleted from object {1} in bucket {2}!\n".format(legal_hold_id, object_name, bucket_name))

延長受保護物件的保留期間

只能延長物件的保留期間。 不能從目前配置的值縮短。

保留擴充值以三種方式之一來設定：

• 現行值再加上時間（Additional-Retention-Period 或類似方法）
• 新擴充期間（以秒為單位）（Extend-Retention-From-Current-Time 或類似方法）
• 物件的新保留到期日（New-Retention-Expiration-Date 或類似方法）

物件 meta 資料中所儲存的現行保留期間可透過給定的其他時間增加，或取代為新值，視 extendRetention 要求中所設定的參數而定。 在所有情況下，會針對現行保留期間檢查延長保留參數，而且只有在更新的保留期間大於現行保留期間時，才會接受延長的參數。

不再保留的受保護儲存區物件（保留期間已過期，而物件沒有任何合法保留），在被改寫時會再次保留。 新的保留期間可以提供為物件改寫要求的一部分，否則會將儲存區的預設保留時間提供給物件。

def extend_retention_period_on_object(bucket_name, object_name, additional_seconds):
    print("Extend the retention period on {0} in bucket {1} by {2} seconds.\n".format(object_name, bucket_name, additional_seconds))

    cos_client.extend_object_retention(
        Bucket=bucket_ame,
        Key=object_name,
        AdditionalRetentionPeriod=additional_seconds
    )

    print("New retention period on {0} is {1}\n".format(object_name, additional_seconds))

列出受保護物件的合法保留

此作業傳回：

• 物件建立日期
• 物件保留期間（秒）
• 根據期間和建立日期計算的保留到期日
• 合法保留的清單
• 合法保留 ID
• 套用合法保留時的時間戳記

如果物件沒有任何合法保留，則會傳回空的 LegalHoldSet。 如果在物件上未指定保留期間，則會傳回 404 錯誤。

def list_legal_holds_on_object(bucket_name, object_name):
    print("List all legal holds on object {0} in bucket {1}\n".format(object_name, bucket_name));

    response = cos_client.list_legal_holds(
        Bucket=bucket_name,
        Key=object_name
    )

    print("Legal holds on bucket {0}: {1}\n".format(bucket_name, response))

建立代管的靜態網站

這項作業需要許可權，因為通常只允許儲存區擁有者配置儲存區來管理靜態網站。 這些參數決定網站訪客的預設字尾，以及選用的錯誤文件。

def putBucketWebsiteConfiguration(bucket_name):
    website_defaults = {
        'ErrorDocument': {'Key': 'error.html'},
        'IndexDocument': {'Suffix': 'index.html'},
    }
    cos_client.put_bucket_website(Bucket=bucket_name, WebsiteConfiguration=website_defaults)
    print("Website configuration set on bucket {0}\n".format(bucket_name))