def generate_dynamodb_tables():
    """Create the Blockade DynamoDB tables."""
    logger.debug("[#] Setting up DynamoDB tables")
    client = boto3.client('dynamodb', region_name=PRIMARY_REGION)
    existing_tables = client.list_tables()['TableNames']

    responses = list()
    for label in DYNAMODB_TABLES:
        if label in existing_tables:
            logger.debug("[*] Table %s already exists" % (label))
            continue

        kwargs = {
            'TableName': label,
            'ProvisionedThroughput': {
                'ReadCapacityUnits': 5,
                'WriteCapacityUnits': 5
            }
        }
        kwargs.update(DYNAMODB_SCHEMAS[label])
        response = client.create_table(**kwargs)
        responses.append(response)
        logger.debug("[#] Successfully setup DynamoDB table %s" % (label))
    logger.info("[#] Successfully setup DynamoDB tables")

    return responses
def generate_dynamodb_tables ( ) : logger . debug ( "[#] Setting up DynamoDB tables" ) client = boto3 . client ( 'dynamodb' , region_name = PRIMARY_REGION ) existing_tables = client . list_tables ( ) [ 'TableNames' ] responses = list ( ) for label in DYNAMODB_TABLES : if label in existing_tables : logger . debug ( "[*] Table %s already exists" % ( label ) ) continue kwargs = { 'TableName' : label , 'ProvisionedThroughput' : { 'ReadCapacityUnits' : 5 , 'WriteCapacityUnits' : 5 } } kwargs . update ( DYNAMODB_SCHEMAS [ label ] ) response = client . create_table ( ** kwargs ) responses . append ( response ) logger . debug ( "[#] Successfully setup DynamoDB table %s" % ( label ) ) logger . info ( "[#] Successfully setup DynamoDB tables" ) return responses
Create the Blockade DynamoDB tables.