Function create_bucket

awscli/testutils.py:176–217

Creates a bucket :returns: the name of the bucket created

(session, name=None, region=None)

def create_bucket(session, name=None, region=None):
    """
    Creates a bucket
    :returns: the name of the bucket created
    """
    if not region:
        region = 'us-west-2'
    client = session.create_client('s3', region_name=region)
    if name:
        bucket_name = name
    else:
        bucket_name = random_bucket_name()
    params = {'Bucket': bucket_name, 'ObjectOwnership': 'ObjectWriter'}
    if region != 'us-east-1':
        params['CreateBucketConfiguration'] = {'LocationConstraint': region}
    try:
        client.create_bucket(**params)
    except ClientError as e:
        if e.response['Error'].get('Code') == 'BucketAlreadyOwnedByYou':
            # This can happen in the retried request, when the first one
            # succeeded on S3 but somehow the response never comes back.
            # We still got a bucket ready for test anyway.
            pass
        else:
            raise
    client.put_bucket_encryption(
        Bucket=bucket_name,
        ServerSideEncryptionConfiguration={
            'Rules': [
                {
                    'ApplyServerSideEncryptionByDefault': {
                        'SSEAlgorithm': 'AES256',
                    },
                    'BlockedEncryptionTypes': {
                        'EncryptionType': ['NONE'],
                    },
                }
            ],
        },
    )
    client.delete_public_access_block(Bucket=bucket_name)
    return bucket_name


def create_dir_bucket(session, name=None, location=None):