Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Update or create a CloudFormation stack given a name and template + params'
'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:])
@kjenney

This comment has been minimized.

Copy link

commented May 29, 2018

Thanks!

@n1mh

This comment has been minimized.

Copy link

commented Jun 18, 2018

Hello,

great script. Is it possible to paste the parameters file? I've been having troubles with that part and it'd be very useful.

Thanks!

@daparthi001

This comment has been minimized.

Copy link

commented Jun 27, 2018

Can you lets us know how to pass parameters file

@gjoris

This comment has been minimized.

Copy link

commented Jun 28, 2018

It's just the filename with the parameters, in the format stated here: https://boto3.readthedocs.io/en/latest/reference/services/cloudformation.html#CloudFormation.Client.update_stack

So, a list with objects, like this:

[
        {
            'ParameterKey': 'string',
            'ParameterValue': 'string',
            'UsePreviousValue': True|False,
            'ResolvedValue': 'string'
        }
]
@daparthi001

This comment has been minimized.

Copy link

commented Jul 5, 2018

"main",
"template_data = _parse_template(template_data)"
],
[
"/var/task/lambda_function.py",
64,
"_parse_template",
"with open(template) as template_fileobj:"
]
],
"errorType": "TypeError",
"errorMessage": "coercing to Unicode: need string or buffer, dict found"
}

Request ID:
"bc631cf4-806e-11e8-b9f2-fb0aad2a39fd"

Function Logs:
START RequestId: bc631cf4-806e-11e8-b9f2-fb0aad2a39fd Version: $LATEST
coercing to Unicode: need string or buffer, dict found: TypeError
Traceback (most recent call last):
File "/var/task/lambda_function.py", line 18, in lambda_handler
main(event['stack_name'], event['template_bucket'], event['template_key'], event['Parameters'])
File "/var/task/lambda_function.py", line 27, in main
template_data = _parse_template(template_data)
File "/var/task/lambda_function.py", line 64, in _parse_template
with open(template) as template_fileobj:
TypeError: coercing to Unicode: need string or buffer, dict found

Getting below error ?

@abdennour

This comment has been minimized.

Copy link

commented Jul 5, 2018

aws cloudformation deploy

I think you need to check aws cloudformation deploy mentioned in AWS CLI 1.15.51 and above

To update a stack, specify the name of an existing stack. To create a new stack, specify a new stack name.

a

@daparthi001

This comment has been minimized.

Copy link

commented Jul 12, 2018

#MADE Few modifications for the script ,We can use the CFT update from the s3 and passing the parameter file from s3

'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',region_name='us-east-1')  # pylint: disable=C0103
log = logging.getLogger('deploy.cf.create_or_update')  # pylint: disable=C0103

def lambda_handler(event, context):
    cf = boto3.client('cloudformation')  # pylint: disable=C0103
    log = logging.getLogger('deploy.cf.create_or_update')  # pylint: disable=C0103
    main(event['stack_name'], event['template_bucket'], event['template_key'], event['Parameters'])
def main(stack_name, template_bucket, template_key, Parameters):
    s3_client = boto3.client('s3')
    cf = boto3.client('cloudformation') 
    
    template_data = s3_client.get_object(Bucket=template_bucket,Key=template_key)
    parameter_data = s3_client.get_object(Bucket=template_bucket,Key=Parameters)
    template_json_data = template_data['Body'].read(template_data['ContentLength'])
    parameter_json_data = parameter_data['Body'].read(parameter_data['ContentLength'])
    pararm = json.loads(str(parameter_json_data))
    #print (template_json_data)
   # print(parameter_json_data)
    
    template_data = _parse_template(template_data)
    #parameter_data = _parse_parameters(Parameters)
    params = {
        'StackName': stack_name,
        'TemplateBody':template_json_data,
        'Parameters': pararm,
    }
    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_data):
    #with open(template_data) as template_fileobj:
    #template_data = template_fileobj.read()
    cf.validate_template(TemplateURL='LINK FOR S3 BUCKET')
    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:])
@daparthi001

This comment has been minimized.

Copy link

commented Jul 12, 2018

####INPUT Trigger

{
"account": "123456789012",
"region": "us-east-1",
"detail": {},
"detail-type": "Scheduled Event",
"source": "aws.events",
"stack_name": "STACKNAME",
"template_bucket": "BBUCKET LOCATION",
"template_key": "TEMPLATE FILE",
"Parameters": "Parameters.json",
"time": "",
"id": "",
"resources": [
""
]
}

@direvus

This comment has been minimized.

Copy link

commented Aug 21, 2018

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
@svrist

This comment has been minimized.

Copy link
Owner Author

commented Nov 15, 2018

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

@msawangwan

This comment has been minimized.

Copy link

commented Aug 8, 2019

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'])))
@msawangwan

This comment has been minimized.

Copy link

commented Aug 8, 2019

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')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.