Source

awstools / scripts / cfn

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Ludia Inc.
# This software is licensed as described in the file LICENSE, which
# you should have received as part of this distribution.
# Author: Pior Bastida <pbastida@socialludia.com>

import os
import pkg_resources

from argh import arg, alias, ArghParser, confirm, wrap_errors
from argh.exceptions import CommandError
import boto
from boto.exception import BotoServerError

import awstools
from awstools import find_stacks, find_one_stack
from awstools.display import (format_stack_summary,
                              format_stack_events)
from awstools.application import Applications
from awstools import cfntemplate


HELP_SN = "the name of the stack like tt-python-production"
HELP_TMPL = "force a different template file"
HELP_CFG = "path of an alternative configuration file"
HELP_SETTINGS = "path of the application settings configuration file"


def initialize_from_cli(args):
    """
    Read the configuration and settings file and lookup for a stack_info"""

    config = awstools.read_config(args.config)

    if args.settings:
        settings = args.settings
    else:
        settings = config.get("cfn", "settings")

    settings_data = open(os.path.expanduser(settings), 'rb')

    if hasattr(args, 'stack_name'):
        apps = Applications(settings_data)
        app = apps.get(stackname=args.stack_name)
        sinfo = app.get_stack_info_from_stackname(args.stack_name)
    else:
        sinfo = None

    return config, settings, sinfo


def warn_for_live(sinfo):
    if sinfo['live'] and sinfo['Environment'] == 'production':
        if not confirm("WARNING: Updating a live stack! Are you sure? "):
            raise CommandError("Aborted")


@arg('-a', '--all', default=False)
@arg('stack_name', nargs='?', default='')
@alias('list')
def ls(args):
    stacks = find_stacks(args.stack_name, findall=args.all)
    for stack in stacks:
        yield format_stack_summary(stack, oneline=True)


@arg('stack_name', help=HELP_SN)
@arg('--config', default=None, help=HELP_CFG)
@arg('--settings', default=None, help=HELP_SETTINGS)
@arg('--template', help=HELP_TMPL)
def create(args):
    config, settings, sinfo = initialize_from_cli(args)

    # Read template
    template_path = os.path.join(
        config.get("cfn", "templatedir"),
        args.template if args.template else sinfo['template'])
    template = cfntemplate.CfnTemplate(template_path)
    parameters = cfntemplate.CfnParameters(template, sinfo)

    print("\nStack name: {args.stack_name}\n"
          "\nTemplate: {template!r}\n"
          "\nParameters:\n"
          "{parameters!r}\n".format(args=args,
                                    template=template,
                                    parameters=parameters))

    if not confirm('Confirm this creation? ', default=True):
        raise CommandError("Aborted")

    try:
        stackid = boto.connect_cloudformation().create_stack(
            args.stack_name,
            template_body=template.body,
            parameters=parameters,
            capabilities=['CAPABILITY_IAM'])
        print("StackId %s" % stackid)
    except BotoServerError as e:
        if e.error_message:
            raise CommandError("BotoServerError: " + e.error_message)
        else:
            raise e


@arg('stack_name', help=HELP_SN)
@arg('--config', default=None, help=HELP_CFG)
@arg('--settings', default=None, help=HELP_SETTINGS)
@arg('--template', help=HELP_TMPL)
def update(args):
    config, settings, sinfo = initialize_from_cli(args)

    # Read template
    template = cfntemplate.CfnTemplate(
        os.path.join(
            config.get("cfn", "templatedir"),
            args.template if args.template else sinfo['template']
            )
        )

    parameters = cfntemplate.CfnParameters(template, sinfo)

    print("\nStack name: {args.stack_name}\n"
          "\nTemplate: {template!r}\n"
          "\nParameters:\n"
          "{parameters!r}\n".format(args=args,
                                    template=template,
                                    parameters=parameters))
    warn_for_live(sinfo)

    if not confirm('Confirm the update? ', default=True):
        raise CommandError("Aborted")

    try:
        stackid = boto.connect_cloudformation().update_stack(
            args.stack_name,
            template_body=template.body,
            parameters=parameters,
            capabilities=['CAPABILITY_IAM'])
        print("StackId %s" % stackid)
    except BotoServerError as e:
        if e.error_message:
            raise CommandError("BotoServerError: " + e.error_message)
        else:
            raise e


@arg('stack_name', help=HELP_SN)
@arg('--config', default=None, help=HELP_CFG)
@arg('--settings', default=None, help=HELP_SETTINGS)
def delete(args):
    config, settings, sinfo = initialize_from_cli(args)

    stack = find_one_stack(args.stack_name)

    print(format_stack_summary(stack))

    warn_for_live(sinfo)

    if not confirm('Confirm the deletion? ', default=True):
        raise CommandError("Aborted")

    try:
        res = boto.connect_cloudformation().delete_stack(stack.stack_name)
    except BotoServerError as e:
        if e:
            raise CommandError("BotoServerError: " + e.error_message)
        else:
            raise e
    print("Result %s" % res)


@arg('stack_name', help=HELP_SN)
@wrap_errors(ValueError)
def info(args):
    stack = find_one_stack(args.stack_name, summary=False)

    yield format_stack_summary(stack) + '\n'

    for param in stack.parameters:
        yield str(param)
    yield ''

    for output in stack.outputs:
        yield str(output)
    yield ''

    yield format_stack_events(stack, limit=10) + '\n'

    for resource in stack.describe_resources():
        yield "{r}\n  {r.resource_status} {r.physical_resource_id}".format(r=resource)
    yield ''


@arg('stack_name', help=HELP_SN)
@wrap_errors(ValueError)
def resources(args):
    stack = find_one_stack(args.stack_name, summary=False)

    yield format_stack_summary(stack) + '\n'

    tmpl = "  ".join([
                     "{r.logical_resource_id:<24}",
                     "{r.physical_resource_id:<60}",
                     "[{r.resource_status}] {r.resource_type}"
                     ])

    for resource in stack.describe_resources():
        yield tmpl.format(r=resource)
    yield ''


@arg('stack_name', help=HELP_SN)
@wrap_errors(ValueError)
def events(args):
    stack = find_one_stack(args.stack_name, summary=False)
    yield format_stack_summary(stack) + '\n'
    yield format_stack_events(stack) + '\n'


def main():
    parser = ArghParser(version=pkg_resources.get_distribution("awstools").version)
    parser.add_commands([
                        ls,
                        create,
                        update,
                        delete,
                        info,
                        resources,
                        events
                        ])
    parser.dispatch(completion=False)

if __name__ == '__main__':
    main()
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.