def check_for_problem()

in composer_migration/src/composer_migration/lib/strategies/migrate_kubernetes_pod_operator_ast.py [0:0]

• 18 lines of code
• 9 McCabe index (conditional complexity)

    def check_for_problem(file: str) -> dict:
        # read dag in as a string
        with open(file, "r") as f:
            dag = f.read()
        # parse string into abstract syntax tree
        tree = ast.parse(dag)

        # create a traverser that looks for "ast.Assign" nodes
        a = AssignFinder()

        # visit the ast and add all assign nodes to a list
        # we need the "assign" nodes to be able to name the problem variables
        a.visit(tree)

        # go through possible problem notes and check for "affinity" parameter
        problem_nodes = []
        for assign in a.possible_kpo_nodes:
            k = KPOFinder()
            k.visit(assign)
            # assumes only one target ever
            # if target node doesn't have an ID, log a debug statement
            try:
                i = assign.targets[0].id
                if k.has_problem:
                    problem_nodes.append(i)
            except AttributeError:
                logging.debug("Node is not of type with a target")

        output = {"nodes": problem_nodes}
        return output