def describe_meta()

in syndicate/core/resources/dynamo_db_resource.py [0:0]

• 40 lines of code
• 11 McCabe index (conditional complexity)

    def describe_meta(self, name):
        meta = {
            'resource_type': 'dynamodb_table',
            'external': True
        }
        response = self.dynamodb_conn.describe_table(table_name=name)

        if not response:
            return {}

        key_schema = {k['KeyType']: k['AttributeName'] for k in response['KeySchema']}
        attribute_definitions = {k['AttributeName']: k['AttributeType'] for k in response['AttributeDefinitions']}

        hash_key_name = key_schema.get('HASH')
        hash_key_type = attribute_definitions[hash_key_name] if hash_key_name else None

        sort_key_name = key_schema.get('RANGE')
        sort_key_type = attribute_definitions[sort_key_name] if sort_key_name else None

        meta.update({
            'hash_key_name': hash_key_name,
            'hash_key_type': hash_key_type,
            'sort_key_name': sort_key_name,
            'sort_key_type': sort_key_type,
        })

        gsi = response.get('GlobalSecondaryIndexes')
        if gsi:
            global_indexes = []
            for index in gsi:
                key_schema = {k['KeyType']: k['AttributeName'] for k in index['KeySchema']}

                index_key_name = key_schema.get('HASH')
                index_key_type = attribute_definitions.get(index_key_name) if index_key_name else None

                sort_key_name = key_schema.get('RANGE')
                sort_key_type = attribute_definitions.get(sort_key_name) if sort_key_name else None

                index_data = {
                    'name': index.get('IndexName'),
                    'index_key_name': index_key_name,
                    'index_key_type': index_key_type,
                    'index_sort_key_name': sort_key_name,
                    'index_sort_key_type': sort_key_type,
                }
                global_indexes.append(index_data)
            global_indexes.sort(key=lambda k:k['name'])
            meta['global_indexes'] = global_indexes

        return {name: meta}