Simple DynamoDb query using Index and KeyConditionExpression

query.py

ERROR_HELP_STRINGS = {
    # Common Errors
    'InternalServerError': 'Internal Server Error, generally safe to retry with exponential back-off',
    'ProvisionedThroughputExceededException': 'Request rate is too high. If you\'re using a custom retry strategy make sure to retry with exponential back-off.' +
                                              'Otherwise consider reducing frequency of requests or increasing provisioned capacity for your table or secondary index',
    'ResourceNotFoundException': 'One of the tables was not found, verify table exists before retrying',
    'ServiceUnavailable': 'Had trouble reaching DynamoDB. generally safe to retry with exponential back-off',
    'ThrottlingException': 'Request denied due to throttling, generally safe to retry with exponential back-off',
    'UnrecognizedClientException': 'The request signature is incorrect most likely due to an invalid AWS access key ID or secret key, fix before retrying',
    'ValidationException': 'The input fails to satisfy the constraints specified by DynamoDB, fix input before retrying',
    'RequestLimitExceeded': 'Throughput exceeds the current throughput limit for your account, increase account level throughput before retrying',
}

def create_dynamodb_client(region="us-east-1"):
    return boto3.client("dynamodb", region_name=region)


def create_query_input():
    return {
        "TableName": "TABLE_NAME_GO_HERE",
        "IndexName": "INDEX_NAME_GO_HERE",
        "KeyConditionExpression": "#b4430 = :b4430",
        "FilterExpression": "#b4431 <> :b4431 And #b4432 <> :b4432",
        "ExpressionAttributeNames": {"#b4430":"FIELD_NAME_GO_HERE","#b4431":"FIELD_NAME_GO_HERE","#b4432":"FIELD_NAME_GO_HERE"},
        "ExpressionAttributeValues": {":b4430": {"S":"FIELD_VALUE_GO_HERE"},":b4431": {"S":"FIELD_VALUE_GO_HERE"},":b4432": {"S":"FIELD_VALUE_GO_HERE"}}
    }


def execute_query(dynamodb_client, input):
    try:
        response = dynamodb_client.query(**input)
        print("Query successful.")
        # Handle response
    except ClientError as error:
        handle_error(error)
    except BaseException as error:
        print("Unknown error while querying: " + error.response['Error']['Message'])


def handle_error(error):
    error_code = error.response['Error']['Code']
    error_message = error.response['Error']['Message']

    error_help_string = ERROR_HELP_STRINGS[error_code]

    print('[{error_code}] {help_string}. Error message: {error_message}'
          .format(error_code=error_code,
                  help_string=error_help_string,
                  error_message=error_message))


def main():
    # Create the DynamoDB Client with the region you want
    dynamodb_client = create_dynamodb_client(region="eu-west-1")

    # Create the dictionary containing arguments for query call
    query_input = create_query_input()

    # Call DynamoDB's query API
    execute_query(dynamodb_client, query_input)


if __name__ == "__main__":
    main()