def check_nodes()

in azext_edge/edge/providers/check/base/node.py [0:0]

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

def check_nodes(as_list: bool = False) -> Dict[str, Any]:
    from ...base import client
    check_manager = CheckManager(check_name="evalClusterNodes", check_desc="Evaluate cluster nodes")
    padding = (0, 0, 0, 8)
    target = "cluster/nodes"
    check_manager.add_target(target_name=target, conditions=["len(cluster/nodes)>=1"])

    try:
        core_client = client.CoreV1Api()
        nodes: V1NodeList = core_client.list_node()
    except ApiException as ae:
        logger.debug(str(ae))
        api_error_text = UNABLE_TO_FETCH_NODES_MSG
        check_manager.add_target_eval(
            target_name=target,
            status=CheckTaskStatus.error.value,
            value=api_error_text,
        )
        check_manager.add_display(
            target_name=target,
            display=Padding(api_error_text, (0, 0, 0, 8)),
        )
    else:
        if not nodes or not nodes.items:
            target_display = Padding(NO_NODES_MSG, padding)
            check_manager.add_target_eval(
                target_name=target, status=CheckTaskStatus.error.value, value=NO_NODES_MSG
            )
            check_manager.add_display(target_name=target, display=target_display)
            return check_manager.as_dict()

        check_manager.add_target_eval(target_name=target, status=CheckTaskStatus.success.value, value={"len(cluster/nodes)": len(nodes.items)})
        table = _generate_node_table(check_manager, nodes)

        check_manager.add_display(target_name=target, display=Padding("Node Resources", padding))
        check_manager.add_display(target_name=target, display=Padding(table, padding))

    return check_manager.as_dict(as_list)