def from_dict()

in src/hpc/autoscale/node/node.py [0:0]
58 lines of code
11 McCabe index (conditional complexity)

    def from_dict(d: Dict) -> "Node":
        hostname = ht.Hostname(d["hostname"])
        name = ht.NodeName(d["name"])
        node_id = ht.NodeId(d["node-id"])
        nodearray = ht.NodeArrayName(d["nodearray"])
        vm_size = ht.VMSize(d["vm-size"])
        location = ht.Location(d["location"])
        aux_info = vm_sizes.get_aux_vm_size_info(location, vm_size)
        job_ids = d.get("job-ids", [])

        resources = ht.ResourceDict(d.get("resources", {}))
        available = ht.ResourceDict(d.get("available", {}))

        for key, value in list(resources.items()):
            if not isinstance(value, str):
                continue
            if value.startswith("size::"):
                resources[key] = ht.Size.value_of(value)
            elif value.startswith("memory::"):
                resources[key] = ht.Memory.value_of(value)

        for key, value in list(available.items()):
            if not isinstance(value, str):
                continue
            if value.startswith("size::"):
                available[key] = ht.Size.value_of(value)
            elif value.startswith("memory::"):
                available[key] = ht.Memory.value_of(value)
        if d.get("memory"):
            memory: ht.Memory = ht.Memory.value_of(d["memory"])  # type: ignore
        else:
            memory = aux_info.memory
        ret = Node(
            node_id=DelayedNodeId(name, node_id),
            name=name,
            nodearray=nodearray,
            bucket_id=ht.BucketId(d["bucket-id"]),
            hostname=hostname,
            private_ip=ht.IpAddress(d["private-ip"]),
            instance_id=ht.InstanceId(d["instance-id"]),
            vm_size=vm_size,
            location=location,
            spot=d.get("spot", False),
            vcpu_count=d.get("vcpu-count", aux_info.vcpu_count),
            memory=memory,
            infiniband=aux_info.infiniband,
            state=ht.NodeStatus(d.get("state", "unknown")),
            target_state=ht.NodeStatus(d.get("state", "unknown")),
            power_state=ht.NodeStatus("on"),
            exists=d.get("exists", True),
            placement_group=d.get("placement_group"),
            managed=d.get("managed", False),
            resources=resources,
            software_configuration=d.get("software-configuration"),
            keep_alive=d.get("keep-alive", False),
        )

        for job_id in job_ids:
            ret.assign(job_id)

        ret.available.update(d.get("available", {}))
        ret.metadata.update(d.get("metadata", {}))

        return ret