Skip to content

Commit

Permalink
add type hints and function descriptions
Browse files Browse the repository at this point in the history
  • Loading branch information
mandar242 committed Sep 4, 2024
1 parent 7862e9e commit acc5fa8
Showing 1 changed file with 84 additions and 10 deletions.
94 changes: 84 additions & 10 deletions plugins/modules/elb_classic_lb_info.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,12 @@
vpc_id: vpc-c248fda4
"""

from typing import Any
from typing import List
from typing import Dict
from typing import Tuple
from typing import Union

try:
import botocore
except ImportError:
Expand All @@ -136,13 +142,20 @@
from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry
from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_list_to_ansible_dict

from ansible_collections.community.aws.plugins.module_utils.modules import AnsibleCommunityAWSModule as AnsibleAWSModule
from ansible_collections.amazon.aws.plugins.module_utils.modules import AnsibleAWSModule


MAX_AWS_RETRIES = 5
MAX_AWS_DELAY = 5
def list_elbs(connection: Any, load_balancer_names: List[str]) -> List[Dict]:
"""
List Elastic Load Balancers (ELBs) and their detailed information.
Parameters:
connection (boto3.client): The Boto3 ELB client object.
load_balancer_names (List[str]): List of ELB names to gather information about.
def list_elbs(connection, load_balancer_names):
Returns:
A list of dictionaries where each dictionary contains informtion about one ELB.
"""
results = []

if not load_balancer_names:
Expand All @@ -157,7 +170,17 @@ def list_elbs(connection, load_balancer_names):
return results


def describe_elb(connection, lb):
def describe_elb(connection: Any, lb: Dict) -> Dict:
"""
Describes an Elastic Load Balancer (ELB).
Parameters:
connection (boto3.client): The Boto3 ELB client object.
lb (Dict): Dictionary containing ELB .
Returns:
A dictionary with detailed information of the ELB.
"""
description = camel_dict_to_snake_dict(lb)
name = lb["LoadBalancerName"]
instances = lb.get("Instances", [])
Expand All @@ -177,11 +200,30 @@ def describe_elb(connection, lb):

@AWSRetry.jittered_backoff()
def get_all_lb(connection):
"""
Get paginated result for information of all Elastic Load Balancers.
Parameters:
connection (boto3.client): The Boto3 ELB client object.
Returns:
A list of dictionaries containing descriptions of all ELBs.
"""
paginator = connection.get_paginator("describe_load_balancers")
return paginator.paginate().build_full_result()["LoadBalancerDescriptions"]


def get_lb(connection, load_balancer_name):
def get_lb(connection: Any, load_balancer_name: str) -> Union[Dict, List]:
"""
Describes a specific Elastic Load Balancer (ELB) by name.
Parameters:
connection (boto3.client): The Boto3 ELB client object.
load_balancer_name (str): Name of the ELB to gather information about.
Returns:
A dictionary with detailed information of the specified ELB.
"""
try:
return connection.describe_load_balancers(aws_retry=True, LoadBalancerNames=[load_balancer_name])[
"LoadBalancerDescriptions"
Expand All @@ -190,21 +232,53 @@ def get_lb(connection, load_balancer_name):
return []


def get_lb_attributes(connection, load_balancer_name):
def get_lb_attributes(connection: Any, load_balancer_name: str) -> Dict:
"""
Retrieves attributes of specific Elastic Load Balancer (ELB) by name.
Parameters:
connection (boto3.client): The Boto3 ELB client object.
load_balancer_name (str): Name of the ELB to gather information about.
Returns:
A dictionary with detailed information of the attributes of specified ELB.
"""
attributes = connection.describe_load_balancer_attributes(aws_retry=True, LoadBalancerName=load_balancer_name).get(
"LoadBalancerAttributes", {}
)
return camel_dict_to_snake_dict(attributes)


def get_tags(connection, load_balancer_name):
def get_tags(connection: Any, load_balancer_name: str) -> Dict:
"""
Retrieves tags of specific Elastic Load Balancer (ELB) by name.
Parameters:
connection (boto3.client): The Boto3 ELB client object.
load_balancer_name (str): Name of the ELB to gather information about.
Returns:
A dictionary of tags associated with the specified ELB.
"""
tags = connection.describe_tags(aws_retry=True, LoadBalancerNames=[load_balancer_name])["TagDescriptions"]
if not tags:
return {}
return boto3_tag_list_to_ansible_dict(tags[0]["Tags"])


def lb_instance_health(connection, load_balancer_name, instances, state):
def lb_instance_health(connection: Any, load_balancer_name: str, instances: List[Dict], state: str) -> Tuple[List, int]:
"""
Describes the health status of instances associated with a specific Elastic Load Balancer (ELB).
Parameters:
connection (Any): The Boto3 client object for ELB.
load_balancer_name (str): The name of the ELB.
instances (List[Dict]): List of dictionaries containing instances associated with the ELB.
state (str): The health state to filter by (e.g., "InService", "OutOfService", "Unknown").
Returns:
Tuple[List, int]: A tuple containing a list of instance IDs matching state and the count of matching instances.
"""
instance_states = connection.describe_instance_health(LoadBalancerName=load_balancer_name, Instances=instances).get(
"InstanceStates", []
)
Expand All @@ -222,7 +296,7 @@ def main():
)

connection = module.client(
"elb", retry_decorator=AWSRetry.jittered_backoff(retries=MAX_AWS_RETRIES, delay=MAX_AWS_DELAY)
"elb", retry_decorator=AWSRetry.jittered_backoff(retries=5, delay=5)
)

try:
Expand Down

0 comments on commit acc5fa8

Please sign in to comment.