check_dynamodb.py

import boto3
import json


class DynamodbCapacityChecker(object):
    def __init__(self):
        self.client = boto3.client('dynamodb', region_name='ap-northeast-1')

    def check(self) -> None:
        tables = self.get_dynamodb_tables()

        for table_name in tables:
            table_info = self.get_table_info(table_name)
            result = self.parse_table_info(table_info)

            print(json.dumps(result, indent=4))

    def get_dynamodb_tables(self, next_table: str = None) -> list:
        option = {}
        if next_table is not None:
            option['ExclusiveStartTableName'] = next_table
        resp = self.client.list_tables(**option)
        result = resp.get('TableNames', [])

        if 'LastEvaluatedTableName' in resp:
            result += self.get_dynamodb_tables(next_table=resp['LastEvaluatedTableName'])
        return result

    def get_table_info(self, table_name: str) -> dict:
        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.describe_table
        resp = self.client.describe_table(TableName=table_name)
        return resp['Table']

    def parse_table_info(self, table_info: dict) -> dict:
        global_secondary_indexes = []

        if 'GlobalSecondaryIndexes' in table_info:
            for gsi in table_info['GlobalSecondaryIndexes']:
                global_secondary_indexes.append({
                    'index_name': gsi['IndexName'],
                    'rcu': gsi['ProvisionedThroughput']['ReadCapacityUnits'],
                    'wcu': gsi['ProvisionedThroughput']['WriteCapacityUnits'],
                })

        if 'BillingModeSummary' in table_info:
            billing_mode = table_info['BillingModeSummary']['BillingMode']
        elif table_info['ProvisionedThroughput']['ReadCapacityUnits'] != 0 and \
                table_info['ProvisionedThroughput']['WriteCapacityUnits'] != 0:
            # BillingModeSummary が無く、かつ、RCUとWCUが0以外の場合は、PROVISIONEDと見なす
            billing_mode = 'PROVISIONED'
        else:
            billing_mode = '???'

        return {
            'table_name': table_info['TableName'],
            'table_size_bytes': table_info['TableSizeBytes'],
            'item_count': table_info['ItemCount'],
            'billing_mode': billing_mode,
            'rcu': table_info['ProvisionedThroughput']['ReadCapacityUnits'],
            'wcu': table_info['ProvisionedThroughput']['WriteCapacityUnits'],
            'global_secondary_indexes': global_secondary_indexes
        }


if __name__ == '__main__':
    DynamodbCapacityChecker().check()