def get_or_create_default_bucket()

in src/smexperiments/_utils.py [0:0]

• 23 lines of code
• 8 McCabe index (conditional complexity)

def get_or_create_default_bucket(boto_session, default_bucket_prefix="sagemaker"):
    """Creates a default bucket if not already exists. The bucket name is a combination of a prefix, the region, and
    account.

    Args:
        boto_session (boto3.Session): boto session
        default_bucket_prefix (str): prefix to the bucket name

    Returns:
        str: The default bucket name.
    """
    account = boto_session.client("sts").get_caller_identity()["Account"]
    region = boto_session.region_name
    default_bucket = "{}-{}-{}".format(default_bucket_prefix, region, account)

    s3 = boto_session.resource("s3")
    try:
        # 'us-east-1' cannot be specified because it is the default region:
        if region == "us-east-1":
            s3.create_bucket(Bucket=default_bucket)
        else:
            s3.create_bucket(Bucket=default_bucket, CreateBucketConfiguration={"LocationConstraint": region})

    except botocore.exceptions.ClientError as e:
        error_code = e.response["Error"]["Code"]
        message = e.response["Error"]["Message"]
        logging.debug("Create Bucket failed. error code: {}, message: {}".format(error_code, message))

        if error_code == "BucketAlreadyOwnedByYou":
            pass
        elif error_code == "OperationAborted" and "conflicting conditional operation" in message:
            # If this bucket is already being concurrently created, we don't need to create it again.
            pass
        elif error_code == "TooManyBuckets":
            # Succeed if the default bucket exists
            s3.meta.client.head_bucket(Bucket=default_bucket)
        else:
            raise
    return default_bucket