def __init__()

in src/cfnlint/graph.py [0:0]

• 59 lines of code
• 34 McCabe index (conditional complexity)

    def __init__(self, cfn):
        """Builds a graph where resources are nodes and edges are explicit (DependsOn) or implicit (Fn::GetAtt, Fn::Sub, Ref)
        relationships between resources"""

        # Directed graph that allows self loops and parallel edges
        self.graph = networkx.MultiDiGraph(name='template')

        # add all resources in the template as nodes
        for resourceId, resourceVals in cfn.template.get('Resources', {}).items():
            type_val = resourceVals.get('Type', '')
            graph_label = str.format('{0}\\n<{1}>', resourceId, type_val)
            self.graph.add_node(resourceId, label=graph_label)
            target_ids = resourceVals.get('DependsOn', [])
            if isinstance(target_ids, (list, six.string_types)):
                if isinstance(target_ids, (six.string_types)):
                    target_ids = [target_ids]
                for target_id in target_ids:
                    if isinstance(target_id, six.string_types):
                        if self._is_resource(cfn, target_id):
                            target_resource_id = target_id
                            self.graph.add_edge(resourceId, target_resource_id, label='DependsOn')

        # add edges for "Ref" tags. { "Ref" : "logicalNameOfResource" }
        refs_paths = cfn.search_deep_keys('Ref')
        for ref_path in refs_paths:
            ref_type, source_id = ref_path[:2]
            target_id = ref_path[-1]
            if not ref_type == 'Resources':
                continue

            if isinstance(target_id, (six.text_type, six.string_types, int)) and (self._is_resource(cfn, target_id)):
                target_resource_id = target_id
                self.graph.add_edge(source_id, target_resource_id, label='Ref')

        # add edges for "Fn::GetAtt" tags.
        # { "Fn::GetAtt" : [ "logicalNameOfResource", "attributeName" ] } or { "!GetAtt" : "logicalNameOfResource.attributeName" }
        getatt_paths = cfn.search_deep_keys('Fn::GetAtt')
        for getatt_path in getatt_paths:
            ref_type, source_id = getatt_path[:2]
            value = getatt_path[-1]
            if not ref_type == 'Resources':
                continue

            if isinstance(value, list) and len(value) == 2 and (self._is_resource(cfn, value[0])):
                target_resource_id = value[0]
                self.graph.add_edge(source_id, target_resource_id, label='GetAtt')

            if isinstance(value, (six.string_types, six.text_type)) and '.' in value:
                target_resource_id = value.split('.')[0]
                if self._is_resource(cfn, target_resource_id):
                    self.graph.add_edge(source_id, target_resource_id, label='GetAtt')

        # add edges for "Fn::Sub" tags. E.g. { "Fn::Sub": "arn:aws:ec2:${AWS::Region}:${AWS::AccountId}:vpc/${vpc}" }
        sub_objs = cfn.search_deep_keys('Fn::Sub')
        for sub_obj in sub_objs:
            sub_parameters = []
            sub_parameter_values = {}
            value = sub_obj[-1]
            ref_type, source_id = sub_obj[:2]

            if not ref_type == 'Resources':
                continue

            if isinstance(value, list):
                if not value:
                    continue
                if len(value) == 2:
                    sub_parameter_values = value[1]
                sub_parameters = self._find_parameter(value[0])
            elif isinstance(value, (six.text_type, six.string_types)):
                sub_parameters = self._find_parameter(value)

            for sub_parameter in sub_parameters:
                if sub_parameter not in sub_parameter_values:
                    if '.' in sub_parameter:
                        sub_parameter = sub_parameter.split('.')[0]
                    if self._is_resource(cfn, sub_parameter):
                        self.graph.add_edge(source_id, sub_parameter, label='Sub')