def create_update_stack()

in lambdas/custom_resources/CTE_CrossAccountCloudFormation/src/cfn_helper.py [0:0]

• 55 lines of code
• 7 McCabe index (conditional complexity)

def create_update_stack(stack_name, template, cfn_params, capability, region='us-east-1', waiter=False, tags=None, session=None):
    """Creates or updates a cloudformation stack using the provided parameters and
    optionally waits for it to be complete

    Args:
        stack_name (str): Name of the stack to create/update
        template (str or dict): Path to the template file on disk to create stack with
        cfn_params (list of dict, optional): List of parameter structures that specify input parameters for the stack
        capability (str): The capability string noting if the stack contains IAM or custom named IAM resources
                            Options: 'CAPABILITY_IAM'|'CAPABILITY_NAMED_IAM'
        region (str): AWS Region
        waiter (bool): True/False if we should wait for the stack to complete or immediately return response
        tags (list): tags set on CloudFormation stack
        session (object, optional): boto3 session object

    Returns:
        dict: Standard AWS response dict
    """
    # Setup default arguments for cfn
    args = dict()
    args['StackName'] = stack_name
    args['Capabilities'] = list()
    args['session'] = session

    # Setup CloudFormation Path and/or Body
    args['TemplateBody'] = json.dumps(template)
    template_body = args['TemplateBody']

    # Does CloudFormation Stack already Exists
    stack_exists = describe_stack(
        stack_name=stack_name,
        session=session
    )

    # Setup Tags
    if tags:
        args['Tags'] = tags
    if not tags and stack_exists:
        args['Tags'] = stack_exists['Stacks'][0]['Tags']

    # If CloudFormation stack is currently in progress pickup where it was left off
    in_progress_status = [
        "CREATE_IN_PROGRESS",
        "UPDATE_IN_PROGRESS",
        "DELETE_IN_PROGRESS",
        "ROLLBACK_IN_PROGRESS",
        "UPDATE_COMPLETE_CLEANUP_IN_PROGRESS",
        "UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS",
        "UPDATE_ROLLBACK_IN_PROGRESS"
    ]

    if stack_exists and (stack_exists['Stacks'][0]['StackStatus'] in in_progress_status):
        wait_all_stacks([{"Name": stack_name, "Session": session}])
        return

    # Setup capability to be a list so CloudFormation doesn't fail
    if isinstance(capability, str):
        args['Capabilities'] = [capability]
    elif isinstance(capability, list):
        args['Capabilities'] = capability

    # Check to see if an Update or Create needs to occur
    if stack_exists:
        # if stack exists get integration existing / override parameters
        if stack_exists['Stacks'][0].get('Parameters'):
            logger.info(f"Parameters found in existing Cfn Stack ({stack_name})")
            parameters = update_parameters(
                override_parameters=cfn_params,
                current_parameters=stack_exists['Stacks'][0]['Parameters']
            )

        else:
            logger.info(f"No Parameters found in existing Cfn Stack ({stack_name})")
            parameters = update_parameters(override_parameters=cfn_params)

        args['Parameters'] = remove_unused_parameters(template=template_body, parameters=parameters)
        response = update_stack(**args)
        cfn_action = 'stack_update_complete'

    else:
        args['Parameters'] = update_parameters(override_parameters=cfn_params)
        logger.info(f"Parameters:{args['Parameters']}")
        response = create_stack(**args)
        cfn_action = 'stack_create_complete'

    # Check to see if a waiter is needed and a proper response was received
    if response and waiter:
        stack_id = response['StackId']
        stack_url = f"https://console.aws.amazon.com/cloudformation/home?region={region}#/stack/detail?stackId={re.sub('/', '%2F', stack_id)}"
        logger.info("Waiting for CloudFormation Stack to complete")
        wait_for_stack_complete(stack_name=stack_name, stack_url=stack_url, cfn_action=cfn_action, session=session)

    return response