Update or create a CloudFormation stack given a name and template + params'

cf_create_or_update.py

'Update or create a stack given a name and template + params'
from __future__ import division, print_function, unicode_literals

from datetime import datetime
import logging
import json
import sys

import boto3
import botocore

cf = boto3.client('cloudformation')  # pylint: disable=C0103
log = logging.getLogger('deploy.cf.create_or_update')  # pylint: disable=C0103


def main(stack_name, template, parameters):
    'Update or create stack'

    template_data = _parse_template(template)
    parameter_data = _parse_parameters(parameters)

    params = {
        'StackName': stack_name,
        'TemplateBody': template_data,
        'Parameters': parameter_data,
    }

    try:
        if _stack_exists(stack_name):
            print('Updating {}'.format(stack_name))
            stack_result = cf.update_stack(**params)
            waiter = cf.get_waiter('stack_update_complete')
        else:
            print('Creating {}'.format(stack_name))
            stack_result = cf.create_stack(**params)
            waiter = cf.get_waiter('stack_create_complete')
        print("...waiting for stack to be ready...")
        waiter.wait(StackName=stack_name)
    except botocore.exceptions.ClientError as ex:
        error_message = ex.response['Error']['Message']
        if error_message == 'No updates are to be performed.':
            print("No changes")
        else:
            raise
    else:
        print(json.dumps(
            cf.describe_stacks(StackName=stack_result['StackId']),
            indent=2,
            default=json_serial
        ))


def _parse_template(template):
    with open(template) as template_fileobj:
        template_data = template_fileobj.read()
    cf.validate_template(TemplateBody=template_data)
    return template_data


def _parse_parameters(parameters):
    with open(parameters) as parameter_fileobj:
        parameter_data = json.load(parameter_fileobj)
    return parameter_data


def _stack_exists(stack_name):
    stacks = cf.list_stacks()['StackSummaries']
    for stack in stacks:
        if stack['StackStatus'] == 'DELETE_COMPLETE':
            continue
        if stack_name == stack['StackName']:
            return True
    return False


def json_serial(obj):
    """JSON serializer for objects not serializable by default json code"""
    if isinstance(obj, datetime):
        serial = obj.isoformat()
        return serial
    raise TypeError("Type not serializable")


if __name__ == '__main__':
    main(*sys.argv[1:])

Great gist @svrist, thanks for posting. Depending on the number of stacks present in the account, your implementation of _stack_exists() may return a false negative, because list_stacks may page the results.

To guarantee a correct result from _stack_exists() you should use a paginator, e.g.,

    paginator = cf.get_paginator('list_stacks')
    for page in paginator.paginate():
        for stack in page['StackSummaries']:
            if stack['StackStatus'] == 'DELETE_COMPLETE':
                continue
            if stack['StackName'] == stack_name:
                return True
    return False

Thanks for all the replies. Great points on the pagination @direvus

pro tip: if you're sick of writing that serializer function in all your boto3 code, just pass str to json.dumps, ie:

print(json.dumps(o, indent=2, default=str))

if i find myself writing this too many times in a file then, i'll make a lambda (some where appropriate of course):

dmp = lambda o: json.dumps(o, indent=2, default=str)

...

print(dmp(cf.describe_stacks(StackName=stack_result['StackId'])))

oh and to add to @direvus comment regarding pagination.. you can also cutdown on:

a) the if/else status check
b) the number of stack summaries returned

by utilizing the StackStatusFilter function parameter. also, i tend to pass the client around rather than storing a global, ie:

def stacks_by_status(cf, status_include_filter):
    """
    ``status_include_filter`` should be a list ...
    """

    pages = cf.get_paginator('list_stacks').paginate(
        StackStatusFilter=status_include_filter)

    for page in pages:
        for s in page.get('StackSummaries', []):
            yield s

then you can define your exists function using this (or don't even need a function really), ie:

def stack_exists(cf, stack_name):
    for s in stacks_by_status(cf, ['CREATE_COMPLETE', etc ...]):
        if s.get('StackName', '') == stack_name:
            return s

    return None

...

if stack_exists(cf, 'some_stack'):
    print('stack exists')
else:
    print('stack does not exist')

Can we use yaml files?

How Do we use Yaml File in this scenario ?

How Do we use Yaml File in this scenario ?

@dheeraj-thedev @AiswaryaGopal

I am also looking for YAML example using boto3, however, I found troposphere which supports Yaml files. An example is given here.

@illusivedeveloper @AiswaryaGopal
https://boto3.amazonaws.com/v1/documentation/api/1.9.42/reference/services/cloudformation.html
I found this on boto3 official documentation that

**validate_template(kwargs)
Validates a specified template. AWS CloudFormation first checks if the template is valid JSON. If it isn't, AWS CloudFormation checks if the template is valid YAML. If both these checks fail, AWS CloudFormation returns a template validation error.

 def _parse_template(template):
        with open(template) as template_fileobj:
        template_data = template_fileobj.read()
        cf.validate_template(TemplateBody=template_data)
        return template_data

Very helpful, thank you!!

Where to mention our json template file location