Collecte des informations requises

Les variables suivantes apparaissent dans les exemples :

• bucket_name doit être une chaîne unique et sécurisée DNS. Etant donné que les noms de compartiment sont uniques sur l'ensemble du système, ces valeurs devront être modifiées si cet exemple est exécuté plusieurs fois. Notez que les noms sont réservés pendant 10 à 15 minutes après leur suppression.
• ibm_api_key_id est la valeur trouvée dans les données d'identification de service sous la forme apikey.
• ibm_service_instance_id est la valeur trouvée dans les données d'identification de service sous la forme resource_instance_id.
• endpoint_url est une URL de noeud final de service qui englobe le protocole https://. Cette valeur n'est pas la valeur endpoints qui se trouve dans les données d'identification de service. Pour plus d'informations sur les noeuds finaux, voir Noeuds finaux et emplacements de stockage.
• LocationConstraint est un code de mise à disposition valide qui correspond à la valeur endpoint.

Initialisation de la configuration

Cet exemple crée un objet resource. Une ressource fournit une interface orientée objet vers Object Storage. Cela permet un niveau d'abstraction plus élevé que les appels de bas niveau fournis par un objet client.

Notez que certaines opérations (telles que le transfert à grande vitesse d' Aspera ) nécessitent un client objet. Aspera nécessite la version Python d' 3.6.

Avis concernant les versions antérieures: la prise en charge d' Aspera est considérée comme obsolète. Utilisez plutôt le 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
)

Un client fournit une interface de bas niveau vers l'API Object StorageS3. Cela permet de traiter directement les réponses HTTP, plutôt que d’utiliser les méthodes et attributs abstraits fournis par une ressource pour accéder aux informations contenues dans les en-têtes ou les données de réponse 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
)

Création d'un nouveau compartiment

Les exemples ci-dessous utilisent un client qui est une interface de bas niveau.

Une liste de codes de mise à disposition valides pour LocationConstraint peut être référencée dans le guide sur les classes de stockage.

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

Création d'un nouveau fichier texte

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

Création de la liste des compartiments disponibles

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

Création de la liste des éléments contenus dans un compartiment

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

Obtention du contenu de fichier d'un élément particulier

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

Suppression d'un élément d'un compartiment

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

Suppression de plusieurs éléments d'un compartiment

La demande de suppression peut contenir un maximum de 1000 clés. Bien que cela permette de réduire l'impact sur les performances par requête, soyez vigilant lorsque vous supprimez un grand nombre de clés. Tenez compte également de la taille des objets de manière à obtenir de bonnes performances.

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

Suppression d'un compartiment

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

Exécution d'un envoi par téléchargement en plusieurs parties

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

Envoi par téléchargement d'objet volumineux à l'aide de TransferManager

TransferManager représente un autre moyen d'exécuter des transferts de fichiers volumineux en incorporant automatiquement des envois par téléchargement en plusieurs parties chaque fois que cela est nécessaire lors de la définition des paramètres de configuration.

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

Création de la liste des éléments contenus dans un compartiment (v2)

S3.Client du client comporte une méthode mise à jour pour répertorier le contenu (list_objects_v2). Cette méthode vous permet de limiter le nombre d'enregistrements renvoyés et d'extraire les enregistrements par lots. Cela peut s'avérer utile pour paginer vos résultats dans une application et améliorer les performances.

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

Création d'une politique de sauvegarde

# 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 }")

Lister une politique de sauvegarde

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

Mettre en place une politique de sauvegarde

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

Supprimer une politique de sauvegarde

# 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.")

Création d'une chambre forte de sauvegarde

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

Liste des chambres fortes de sauvegarde

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

Obtenir des chambres fortes de sauvegarde

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

Mise à jour des chambres fortes de sauvegarde

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

Supprimer une chambre forte de sauvegarde

# 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}'.")

Plages de récupération des listes

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

Obtenir la plage de récupération

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

Mise à jour de la plage de récupération

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

Lancement d'une restauration

# 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}")

Rétablissement des listes

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

Obtenir les détails de la restauration

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

Créer un nouveau compartiment Object Storage avec le verrouillage d'objet activé

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

Activer la configuration du verrouillage d'objet avec le mode de conformité sur le compartiment Object Storage

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)

Configurer le verrouillage des objets avec le mode de gouvernance sur le compartiment Object Storage

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)

Obtenir la configuration du verrouillage d'objet sur le compartiment Object Storage

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)

Importer un objet en mode de gouvernance vers le compartiment Object Storage

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

Activer le verrouillage de conservation des objets avec le mode de conformité sur l'objet

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)

Activer la conservation du verrouillage d'objet avec le mode de gouvernance sur l'objet

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)

Obtenir la conservation du verrouillage d'objet

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)

Mettre l'objet en verrouillage légal

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

Obtenir le verrouillage légal de l'objet

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)

Suppression d'un objet avec le mode de gouvernance de verrouillage d'objet à l'aide de la gouvernance de contournement

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)

Avant de commencer

Les éléments suivants sont nécessaires pour créer un compartiment avec la fonctionnalité Key-Protect activée :

• Un service Key Protect doit être mis à disposition.
• Une clé racine doit être disponible (générée ou importée).

Extraction du CRN de clé racine

1. Extrayez l'ID d'instance pour votre service Key Protect.
2. Utilisez l'API Key Protect pour extraire toutes vos clés disponibles.
   • Vous pouvez utiliser des commandes curl ou un client REST API, tel que Postman, pour accéder à l'API Key Protect.
3. Extrayez le CRN de la clé racine que vous utiliserez pour activer Key Protect sur votre compartiment. Le CRN se présente comme suit :

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

Création d'un compartiment avec Key Protect activé

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

Initialisation de AsperaTransferManager

Avant d’initialiser le AsperaTransferManager, assurez-vous de disposer d’un objet ( client et non d’un resource ou d’un session) en état de fonctionnement.

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)

Vous devez fournir une clé d'API IAM pour Aspera High-Speed Transfer. Les données d'identification HMAC ne sont PAS actuellement prises en charge. Pour plus d'informations sur IAM, cliquez ici.

Pour obtenir une capacité de traitement la plus élevée possible, fractionnez le transfert en un certain nombre de sessions parallèles qui envoient des blocs de données dont la taille est définie par la valeur threshold.

La configuration standard pour l'utilisation de plusieurs sessions doit être la suivante :

• Débit cible de 2500 Mbit/s
• Seuil de 100 Mo (il s'agit de la valeur recommandée pour la plupart des applications)

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

Dans l'exemple ci-dessus, le SDK génère suffisamment de sessions pour tenter d'atteindre le débit cible correspondant à 2500 Mbit/s.

La gestion de session peut également être configurée explicitement dans le SDK. Cela s'avère utile si vous souhaitez obtenir un contrôle plus précis de l'utilisation du réseau.

La configuration standard pour l'utilisation de plusieurs sessions explicites doit être la suivante :

• 2 ou 10 sessions
• Seuil de 100 Mo (il s'agit de la valeur recommandée pour la plupart des applications)

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)

Pour obtenir les meilleures performances dans la plupart des cas de figure, utilisez toujours plusieurs sessions afin de réduire au minimum le traitement lié à l'instanciation d'un transfert haut débit Aspera. **Si votre capacité réseau est d'au moins 1 Gbit/s, vous devez utiliser 10 sessions. ** Des réseaux avec une bande passante inférieure doivent utiliser 2 sessions.

Envoi par téléchargement d'un fichier

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

Réception par téléchargement d'un fichier

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

Envoi par téléchargement d'un répertoire

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

Réception par téléchargement d'un répertoire

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

Utilisation des abonnés

Les abonnés offrent une certaine observabilité des transferts en associant des méthodes de rappel personnalisées. Tous les transferts passent par les phases suivantes :

Queued - In Progress - Done

Trois abonnés sont disponibles pour chaque phase :

• CallbackOnQueued() - Appelé lorsqu'un nouveau transfert est ajouté à AsperaTransferManager.
• CallbackOnProgress() - Appelé lorsqu'un transfert a transmis des données (déclenché de manière répétée pendant que le transfert est en cours).
• CallbackOnDone() - Appelé lorsque le transfert est terminé.

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!

Pause/Reprise/Annulation

Le SDK permet de gérer la progression des transferts de fichiers/répertoires via les méthodes suivantes de l'objet AsperaTransferFuture :

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

Il n'y a pas d'effets secondaires liés à l'appel de l'une ou l'autre des méthodes ci-dessus. Le nettoyage adéquat est géré par le 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()

Traitement des incidents liés à Aspera

Problème: Les développeurs qui utilisent n'importe quelle version de Python en plus de 3.6 peuvent rencontrer des échecs lors de l'installation ou de l'utilisation de Aspera SDK.

Cause : si différentes versions de Python sont installées sur votre environnement, vous risquez de rencontrer des échecs d'installation lorsque vous essayez d'installer le Aspera SDK. Cela peut être dû à un fichier DLL manquant ou à une DLL incorrecte dans le chemin.

Solution : en premier lieu, réinstallez les bibliothèques Aspera. Il se peut que leur installation ait échoué. Cela peut avoir affecté les fichiers DLL. Si cela ne résout pas les problèmes, vous devez alors mettre à jour votre version d' Python. Si vous ne parvenez pas à le faire, vous pouvez utiliser la distribution Intel® pour Python * lors de l'installation. Cela doit vous permettre d'installer le Aspera SDK sur Python 3.6.x sans aucun problème.

Utilisation de PUT pour mettre à jour les métadonnées

Remarque : la requête PUT écrase le contenu existant de l'objet; celui-ci doit donc d'abord être téléchargé, puis réuploadé avec les nouvelles métadonnées.

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

Utilisation de COPY pour mettre à jour les métadonnées

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

Ajout d'une configuration de protection à un compartiment existant

Les objets écrits dans un compartiment protégé ne peuvent pas être supprimés tant que la période de protection n'est pas arrivée à expiration et que les conservations légales associées à l'objet n'ont pas toutes été retirées. La valeur de conservation par défaut du compartiment est attribuée à un objet, sauf si une valeur spécifique à l'objet est fournie lors de la création de l'objet. Les objets situés dans des compartiments protégés qui ne sont plus soumis à une obligation de conservation (la période de conservation a expiré et l'objet ne fait l'objet d'aucune conservation à des fins juridiques), lorsqu'ils sont écrasés, sont à nouveau soumis à cette obligation. La nouvelle durée de conservation peut être indiquée dans la demande de remplacement de l'objet, ou la durée de conservation par défaut du compartiment est alors appliquée à l'objet.

Les valeurs minimales et maximales prises en charge pour les paramètres de durée de conservation MinimumRetention, DefaultRetention, et sont MaximumRetention respectivement de 0 jour au minimum et de 365 243 jours (1 000 ans) au maximum.

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

Vérification de la protection d'un compartiment

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

Envoi par téléchargement d'un objet protégé

Les objets situés dans des compartiments protégés qui ne sont plus soumis à une obligation de conservation (la période de conservation a expiré et l'objet ne fait l'objet d'aucune conservation à des fins juridiques), lorsqu'ils sont écrasés, sont à nouveau soumis à cette obligation. La nouvelle durée de conservation peut être indiquée dans la demande de remplacement de l'objet, ou la durée de conservation par défaut du compartiment est alors appliquée à l'objet.

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

Ajout ou retrait d'une conservation légale dans un objet protégé

L'objet peut prendre en charge 100 conservations légales :

• Un identificateur de conservation légale est une chaîne comprise entre 1 et 64 caractères. Les caractères valides sont les lettres, les chiffres et les symboles !, _, ., *, (, ) et -.
• Si l’ajout de la conservation légale indiquée porte le nombre total de conservations légales sur l’objet à plus de 100, la nouvelle conservation légale n’est pas ajoutée et une erreur 400 est renvoyée.
• Si un identifiant est trop long, il n’est pas ajouté à l’objet et une erreur 400 est renvoyée.
• Si un identifiant contient des caractères non valides, il n’est pas ajouté à l’objet et une erreur 400 est renvoyée.
• Si un identificateur est déjà en cours d'utilisation sur un objet, la conservation légale existante n'est pas modifiée et la réponse indique que l'identificateur était déjà utilisé et renvoie un code d'erreur 409.
• Si un objet ne comporte pas de métadonnées de durée de conservation, un code d'erreur 400 est renvoyé et l'ajout ou le retrait d'une conservation légale ne sont pas autorisés.

Pour ajouter ou supprimer une mise en attente légale, vous devez disposer des droits Manager pour ce compartiment.

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

Prolongement de la durée de conservation d'un objet protégé

La durée de conservation d'un objet ne peut être que prolongée. La valeur actuellement configurée ne peut pas être diminuée.

La valeur de prolongement de la conservation est définie de l'une des trois façons suivantes :

• En ajoutant une durée supplémentaire par rapport à la valeur en cours (Additional-Retention-Period ou méthode similaire)
• En définissant une nouvelle période de prolongement, exprimée en secondes (Extend-Retention-From-Current-Time ou méthode similaire)
• En définissant une nouvelle date d'expiration de la conservation de l'objet (New-Retention-Expiration-Date ou méthode similaire)

La durée de conservation en cours qui est stockée dans les métadonnées de l'objet est soit augmentée de la durée prolongée indiquée, soit remplacée par la nouvelle valeur, en fonction du paramètre défini dans la demande extendRetention. Dans tous les cas, le paramètre de prolongement de la conservation est vérifié par rapport à la durée de conservation en cours et le paramètre de prolongement n'est accepté que si la durée de conservation mise à jour est supérieure à la durée de conservation en cours.

Les objets situés dans des compartiments protégés qui ne sont plus soumis à une obligation de conservation (la période de conservation a expiré et l'objet ne fait l'objet d'aucune conservation à des fins juridiques), lorsqu'ils sont écrasés, sont à nouveau soumis à cette obligation. La nouvelle durée de conservation peut être indiquée dans la demande de remplacement de l'objet, ou la durée de conservation par défaut du compartiment est alors appliquée à l'objet.

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

Création de la liste des conservations légales associées à un objet protégé

Cette opération renvoie les éléments suivants :

• La date de création de l'objet
• La durée de conservation, exprimée en secondes
• La date d'expiration de la conservation, calculée sur la base de la durée de conservation et de la date de création
• La liste des conservations légales
• L'identificateur de conservation légale
• L'horodatage correspondant à l'application de la conservation légale

Si aucune conservation légale n'est associée à l'objet, un élément LegalHoldSet vide est renvoyé. Si aucune durée de conservation n'est spécifiée pour l'objet, un code d'erreur 404 est renvoyé.

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

Créer un site Web statique hébergé

Cette opération requiert des droits d'accès, car seul le propriétaire du compartiment est généralement autorisé à configurer un compartiment pour héberger un site Web statique. Les paramètres déterminent le suffixe par défaut pour les visiteurs du site ainsi qu'un document d'erreur facultatif.

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