def validate_ecs_run_task_info()

in src/helper.py [0:0]

• 54 lines of code
• 11 McCabe index (conditional complexity)

def validate_ecs_run_task_info(ecs_info: dict):
    """Used to validate that SQS Message contains all necessary information to run the ECS Task

    Args:
        ecs_info (int): ECS Information

    Returns:
        :obj:`json`: Returns json object (dict) of the ECS information
    """
    logger.info("Validating SQS Message contains all the needed information to execution ECS Task")
    required_net_keys = {
        "subnets": ecs_info.get('networkConfiguration', {}).get('awsvpcConfiguration', {}).get('subnets'),
        "securityGroups": ecs_info.get('networkConfiguration', {}).get('awsvpcConfiguration', {}).get('securityGroups')
    }
    required_definition_keys = {
        "container_name": ecs_info.get('overrides', {}).get('containerOverrides', [{}])[0].get('name')
    }

    ecs = boto3.client('ecs')
    logger.info(f"ecs_info:{ecs_info}")

    if None in required_net_keys.values():
        logger.warning("***** REQUIRED NETWORK KEYS NOT PASSED TO RUN TASK *****")
        logger.warning(required_net_keys)

        if ecs_info.get('service') and ecs_info.get('cluster'):
            response = ecs.describe_services(
                cluster=ecs_info['cluster'],
                services=[ecs_info['service']]
            )

            logger.debug(response)
            if not required_net_keys['subnets'] and not required_net_keys['securityGroups']:
                logger.info(f"Updating Networking Configuration:{response['services'][0]['networkConfiguration']}")
                ecs_info['networkConfiguration'] = response['services'][0]['networkConfiguration']

            else:
                if not required_net_keys['subnets']:
                    subnets = response['services'][0]['networkConfiguration']['awsvpcConfiguration']['subnets']
                    logger.info(f"Updating Networking Subnet Configuration:{subnets}")
                    ecs_info['networkConfiguration'] = {"awsvpcConfiguration": {"subnets": subnets}}

                if not required_net_keys['securityGroups']:
                    security_groups = response['services'][0]['networkConfiguration']['awsvpcConfiguration']['securityGroups']
                    logger.info(f"Updating Networking securityGroups Configuration:{security_groups}")
                    ecs_info['networkConfiguration'] = {"awsvpcConfiguration": {"securityGroups": security_groups}}

        else:
            raise Exception(
                "ECS Service and/or Cluster Names did not get passed into ECS Information for "
                "validate_ecs_run_task_info"
            )

    if None in required_definition_keys.values():
        logger.warning("***** REQUIRED DEFINITION KEYS NOT PASSED TO RUN TASK *****")
        logger.warning(required_definition_keys)

        response = ecs.describe_task_definition(
            taskDefinition=ecs_info['taskDefinition']
        )
        logger.debug(response)

        if "overrides" not in ecs_info:
            ecs_info['overrides'] = {}

        if "containerOverrides" not in ecs_info['overrides']:
            ecs_info['overrides']['containerOverrides'] = [{}]

        container_name = response['taskDefinition']['containerDefinitions'][0]['name']
        ecs_info['overrides']['containerOverrides'][0]['name'] = container_name

    # Cleanup of ecs run task dictionary
    logger.info("Removing Service from ecs_info dict")
    del ecs_info['service']
    logger.info(f"ecs_parameters:{ecs_info}")

    return ecs_info