필수 정보 수집

이 예에는 다음 변수가 있습니다.

• bucket_name은 고유하며 DNS를 준수하는 문자열이어야 합니다. 버킷 이름은 전체 시스템에서 고유하므로 이 예가 여러 번 실행되는 경우에는 이러한 값을 변경해야 합니다. 이름은 삭제 후에도 10 - 15분 동안 예약되어 있다는 점을 참고하십시오.
• ibm_api_key_id는 서비스 인증 정보에서 apikey로 찾을 수 있는 값입니다.
• ibm_service_instance_id는 서비스 인증 정보에서 resource_instance_id로 찾을 수 있는 값입니다.
• endpoint_url은 https:// 프로토콜을 포함하는 서비스 엔드포인트 URL입니다. 이 값은 서비스 자격 증명에서 찾을 수 있는 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에 대한 하위 레벨 인터페이스를 제공합니다. 이를 통해 리소스에서 제공하는 추상화된 메서드 및 속성을 사용하여 헤더 또는 XML 응답 페이로드에 포함된 정보에 액세스하는 대신 HTTP 응답을 직접 처리할 수 있습니다.

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()

버킷에 있는 항목 나열(v2)

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))

시작하기 전에

Key Protect가 사용으로 설정된 버킷을 작성하려면 다음 항목이 필요합니다.

• 프로비저닝된 Key Protect 서비스
• 사용 가능한 루트 키(생성 또는 가져오기를 통해 확보)

루트 키 CRN 검색

1. Key Protect 서비스의 인스턴스 ID를 검색하십시오.
2. Key Protect API를 사용하여 사용 가능한 키를 모두 검색하십시오.
   • curl 명령을 사용하거나 Postman과 같은 API REST 클라이언트를 사용하여 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)

Aspera 고속 전송을 위해서는 IAM API 키를 제공해야 합니다. HMAC 신임 정보 는 현재 지원되지 않습니다. IAM에 대한 자세한 정보를 보려면 여기를 클릭하십시오.

처리량을 최고로 높이려면 임계값 값으로 크기가 정의된 데이터 청크를 전송하는 병렬 세션의 지정된 수만큼 전송을 분할하십시오.

다중 세션 사용에 대한 일반적인 구성은 다음과 같습니다.

• 2500MBps의 목표 속도
• 100MB의 임계값(이는 대부분의 애플리케이션에 대해 권장되는 값임)

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

위 예에서 이 SDK는 목표 속도인 2500Mbps에 도달하기 위해 충분한 세션을 생성합니다.

SDK에서 세션 관리를 명시적으로 구성할 수도 있습니다. 이는 네트워크 이용을 더 정확하게 제어해야 하는 경우 유용합니다.

명시적 다중 세션 사용에 대한 일반적인 구성은 다음과 같습니다.

• 2개 또는 10개의 세션
• 100MB의 임계값(이는 대부분의 애플리케이션에 대해 권장되는 값임)

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 고속 전송 인스턴스화와 관련된 모든 처리를 최소화하세요. 네트워크 용량이 1Gbps 이상인 경우에는 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 버전을 업데이트해야 합니다. 이 작업을 수행할 수 없는 경우 Python *용 인텔® 배포 설치를 사용할 수 있습니다. 이를 통해 문제 없이 Python 3.6.x 에 Aspera SDK 를 설치할 수 있습니다.

PUT을 사용한 메타데이터 업데이트

참고: 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를 사용한 메타데이터 업데이트

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 오류와 함께 표시됩니다.
• 오브젝트에 보존 기간 메타데이터가 없는 경우에는 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 또는 이와 유사한 메소드)

오브젝트 메타데이터에 저장된 현재 보존 기간은 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))