modules/backend/lambda-src/utils.py
def annotate_with_instance_state(instances):
    """Get the status of each instance"""
    # Get the region of each instance
    region_to_instances = defaultdict(list)
    for item in instances:
        if "instance_id" in item:
            region_to_instances[item["region"]].append(item["instance_id"])

    # Describe instances per region to get their current state
    state_dict = {}
    for region, instance_ids in region_to_instances.items():
        ec2_client = boto3.client("ec2", region_name=region)
        described_instances = ec2_client.describe_instances(InstanceIds=instance_ids)

        for instance in described_instances.get("Reservations", []):
            if len(instance["Instances"]) > 1:
                raise ValueError(f"More than one instance running: {instance}")
            instance_id = instance["Instances"][0]["InstanceId"]
            state = instance["Instances"][0]["State"]["Name"]

            state_dict[instance_id] = state

    # Annotate the instances with their state
    for item in instances:
        if item.get("instance_id") and item["instance_id"] in state_dict:
            item["state"] = state_dict[item["instance_id"]]

    return instances
Get the status of each instance
Get the status of each instance
Get the status of each instance
def annotate_with_instance_state(instances):
    region_to_instances = defaultdict(list)
    for item in instances:
        if "instance_id" in item:
            region_to_instances[item["region"]].append(item["instance_id"])
    state_dict = {}
    for region, instance_ids in region_to_instances.items():
        ec2_client = boto3.client("ec2", region_name=region)
        described_instances = ec2_client.describe_instances(InstanceIds=instance_ids)
        for instance in described_instances.get("Reservations", []):
            if len(instance["Instances"]) > 1:
                raise ValueError(f"More than one instance running: {instance}")
            instance_id = instance["Instances"][0]["InstanceId"]
            state = instance["Instances"][0]["State"]["Name"]
            state_dict[instance_id] = state
    for item in instances:
        if item.get("instance_id") and item["instance_id"] in state_dict:
            item["state"] = state_dict[item["instance_id"]]
    return instances
def annotate_with_instance_state ( instances ) : region_to_instances = defaultdict ( list ) for item in instances : if "instance_id" in item : region_to_instances [ item [ "region" ] ] . append ( item [ "instance_id" ] ) state_dict = { } for region , instance_ids in region_to_instances . items ( ) : ec2_client = boto3 . client ( "ec2" , region_name = region ) described_instances = ec2_client . describe_instances ( InstanceIds = instance_ids ) for instance in described_instances . get ( "Reservations" , [ ] ) : if len ( instance [ "Instances" ] ) > 1 : raise ValueError ( f"More than one instance running: {instance}" ) instance_id = instance [ "Instances" ] [ 0 ] [ "InstanceId" ] state = instance [ "Instances" ] [ 0 ] [ "State" ] [ "Name" ] state_dict [ instance_id ] = state for item in instances : if item . get ( "instance_id" ) and item [ "instance_id" ] in state_dict : item [ "state" ] = state_dict [ item [ "instance_id" ] ] return instances
Get the status of each instance
Get the status of each instance
"""Get the status of each instance""" # Get the region of each instance # Describe instances per region to get their current state # Annotate the instances with their state
instances