def add_unmanaged_nodes()

in src/hpc/autoscale/node/nodemanager.py [0:0]

• 39 lines of code
• 5 McCabe index (conditional complexity)

    def add_unmanaged_nodes(self, existing_nodes: List[UnmanagedNode]) -> None:

        by_key: Dict[
            Tuple[Optional[ht.PlacementGroup], ht.BucketId], List[Node]
        ] = partition(
            # typing will complain that List[Node] is List[UnmanagedNode]
            # just a limitation of python3's typing
            existing_nodes,  # type: ignore
            lambda n: (n.placement_group, n.bucket_id),
        )

        buckets = partition_single(
            self.__node_buckets, lambda b: (b.placement_group, b.bucket_id)
        )

        for key, nodes_list in by_key.items():
            if key in buckets:
                buckets[key].add_nodes(nodes_list)
                continue

            placement_group, bucket_id = key

            a_node = nodes_list[0]
            # create a null definition, limits and bucket for each
            # unique set of unmanaged nodes
            node_def = NodeDefinition(
                nodearray=ht.NodeArrayName("__unmanaged__"),
                bucket_id=ht.BucketId(bucket_id),
                vm_size=ht.VMSize("unknown"),
                location=ht.Location("unknown"),
                spot=False,
                subnet=ht.SubnetId("unknown"),
                vcpu_count=a_node.vcpu_count,
                gpu_count=a_node.gpu_count,
                memory=a_node.memory,
                placement_group=placement_group,
                resources=ht.ResourceDict(dict(deepcopy(a_node.resources))),
                software_configuration=ImmutableOrderedDict(
                    a_node.software_configuration
                ),
            )

            limits = null_bucket_limits(len(nodes_list), a_node.vcpu_count)

            bucket = NodeBucket(
                node_def, limits, len(nodes_list), nodes_list, artificial=True, valid=True,
            )

            self.__node_buckets.append(bucket)
            bucket.add_nodes(nodes_list)

        self._apply_defaults_all()