create_dynamodb_scaling_table.py

import boto3

def creation_table_scaling(table_name: str, key_name: str, key_type: str, max_read_capacity: int, max_write_capacity: int):
    """
    Create table dynamodb and make it scale then wait for it to be created
    Args:
        table_name (str): the name of the table to create
        key_name (str): the name of the key, and main attribute of the table
        key_type (str): the data type for the attribute, S for string, N for number, B for binary
        max_read_capacity(int) : Max nb of read per minute
        max_write_capacity(int) : Max nb of writes per minute
    Returns:
    """
    dynamodb = boto3.resource('dynamodb')
    scaling_dynamodb = boto3.client('application-autoscaling')
    min_read_capacity = 5
    min_write_capacity = 5
    table = dynamodb.create_table(
        TableName=table_name,
        KeySchema=[
            {
                'AttributeName': key_name,
                'KeyType': 'HASH'
                },
            ],
        AttributeDefinitions=[
            {
                'AttributeName': key_name,
                'AttributeType': key_type
                },
            ],
        ProvisionedThroughput={
            'ReadCapacityUnits': min_read_capacity,
            'WriteCapacityUnits': min_write_capacity
            }
        )
    scalable_dimensions = {'dynamodb:table:ReadCapacityUnits': [min_read_capacity, max_read_capacity],
                           'dynamodb:table:WriteCapacityUnits': [min_write_capacity, max_write_capacity]}
    for scalable_dimension, capacity in scalable_dimensions.items():
        scaling_dynamodb.register_scalable_target(ServiceNamespace="dynamodb",
                                                     ResourceId="table/{}".format(table_name),
                                                     ScalableDimension=scalable_dimension,
                                                     MinCapacity=capacity[0],
                                                     MaxCapacity=capacity[1])
    # create scaling policy, without them the auto scaling will not be applied
    metrics_and_dimension = {'DynamoDBReadCapacityUtilization': 'dynamodb:table:ReadCapacityUnits',
                             'DynamoDBWriteCapacityUtilization': 'dynamodb:table:WriteCapacityUnits'}
    percent_of_use_to_aim_for = 50.0
    scale_out_cooldown_in_seconds = 60
    scale_in_cooldown_in_seconds = 60
    for metric, dimension in metrics_and_dimension.items():
        LOGGER.info(metric)
        LOGGER.info(dimension)
        scaling_dynamodb.put_scaling_policy(ServiceNamespace="dynamodb",
                                               ResourceId="table/{}".format(table_name),
                                               PolicyType='TargetTrackingScaling',
                                               PolicyName="Scale{}".format(metric),
                                               ScalableDimension=dimension,
                                               TargetTrackingScalingPolicyConfiguration={
                                                   'TargetValue': percent_of_use_to_aim_for,
                                                   'PredefinedMetricSpecification': {
                                                       'PredefinedMetricType': metric
                                                   },
                                                   'ScaleOutCooldown': scale_out_cooldown_in_seconds,
                                                   'ScaleInCooldown': scale_in_cooldown_in_seconds
                                               })
    table.meta.client.get_waiter('table_exists').wait(TableName=table_name)