Source code for sceptre.template

# -*- coding: utf-8 -*-


This module implements a Template class, which stores a CloudFormation template
and implements methods for uploading it to S3.

import logging
import threading
import botocore

from sceptre.exceptions import TemplateHandlerNotFoundError
from pkg_resources import iter_entry_points

[docs]class Template(object): """ Template represents an AWS CloudFormation template. It is responsible for loading, storing and optionally uploading local templates for use by CloudFormation. :param name: The name of the template. Should be safe to use in filenames and not contain path segments. :type name: str :param handler_config: The configuration for a Template handler. Must contain a `type`. :type handler_config: dict :param sceptre_user_data: A dictionary of arbitrary data to be passed to\ a handler function in an external Python script. :type sceptre_user_data: dict :param stack_group_config: The StackGroup config for the Stack. :type stack_group_config: dict :param connection_manager: :type connection_manager: sceptre.connection_manager.ConnectionManager :param s3_details: :type s3_details: dict """ _boto_s3_lock = threading.Lock() def __init__( self, name, handler_config, sceptre_user_data, stack_group_config, connection_manager=None, s3_details=None ): self.logger = logging.getLogger(__name__) = name self.handler_config = handler_config if self.handler_config is not None and self.handler_config.get('type') is None: self.handler_config['type'] = 'file' self.sceptre_user_data = sceptre_user_data self.stack_group_config = stack_group_config self.connection_manager = connection_manager self.s3_details = s3_details self._registry = None self._body = None def __repr__(self): return ( "sceptre.template.Template(name='{0}', handler_config={1}, sceptre_user_data={2}, s3_details={3})".format(, self.handler_config, self.sceptre_user_data, self.s3_details ) ) @property def body(self): """ Represents body of the CloudFormation template. :returns: The body of the CloudFormation template. :rtype: str """ if self._body is None: type = self.handler_config.get("type") handler_class = self._get_handler_of_type(type) handler = handler_class(, arguments={k: v for k, v in self.handler_config.items() if k != "type"}, sceptre_user_data=self.sceptre_user_data, connection_manager=self.connection_manager, stack_group_config=self.stack_group_config ) handler.validate() body = handler.handle() if isinstance(body, bytes): body = body.decode('utf-8') if not str(body).startswith("---"): body = "---\n{}".format(body) self._body = body return self._body
[docs] def upload_to_s3(self): """ Uploads the template to ``bucket_name`` and returns its URL. The Template is uploaded with the ``bucket_key``. :returns: The URL of the Template object in S3. :rtype: str :raises: botocore.exceptions.ClientError """ self.logger.debug("%s - Uploading template to S3...", with self._boto_s3_lock: if not self._bucket_exists(): self._create_bucket() # Remove any leading or trailing slashes the user may have added. bucket_name = self.s3_details["bucket_name"] bucket_key = self.s3_details["bucket_key"] bucket_region = self._bucket_region(bucket_name) self.logger.debug( "%s - Uploading template to: 's3://%s/%s'",, bucket_name, bucket_key ) service="s3", command="put_object", kwargs={ "Bucket": bucket_name, "Key": bucket_key, "Body": self.body, "ServerSideEncryption": "AES256" } ) url = "https://{}.s3.{}.amazonaws.{}/{}".format( bucket_name, bucket_region, self._domain_from_region(bucket_region), bucket_key ) self.logger.debug("%s - Template URL: '%s'",, url) return url
def _bucket_exists(self): """ Checks if the bucket ``bucket_name`` exists. :returns: Boolean whether the bucket exists :rtype: bool :raises: botocore.exception.ClientError """ bucket_name = self.s3_details["bucket_name"] self.logger.debug( "%s - Attempting to find template bucket '%s'",, bucket_name ) try: service="s3", command="head_bucket", kwargs={"Bucket": bucket_name} ) except botocore.exceptions.ClientError as exp: if exp.response["Error"]["Message"] == "Not Found": self.logger.debug( "%s - %s bucket not found.",, bucket_name ) return False else: raise self.logger.debug( "%s - Found template bucket '%s'",, bucket_name ) return True def _create_bucket(self): """ Create the s3 bucket ``bucket_name``. :raises: botocore.exception.ClientError """ bucket_name = self.s3_details["bucket_name"] self.logger.debug( "%s - Creating new bucket '%s'",, bucket_name ) if self.connection_manager.region == "us-east-1": service="s3", command="create_bucket", kwargs={"Bucket": bucket_name} ) else: service="s3", command="create_bucket", kwargs={ "Bucket": bucket_name, "CreateBucketConfiguration": { "LocationConstraint": self.connection_manager.region } } )
[docs] def get_boto_call_parameter(self): """ Returns the CloudFormation template location. Uploads the template to S3 and returns the object's URL, or returns the template itself. :returns: The boto call parameter for the template. :rtype: dict """ if self.s3_details: url = self.upload_to_s3() return {"TemplateURL": url} else: return {"TemplateBody": self.body}
def _bucket_region(self, bucket_name): region = service="s3", command="get_bucket_location", kwargs={"Bucket": bucket_name} ).get("LocationConstraint") return region if region else "us-east-1" @staticmethod def _domain_from_region(region): return "" if region.startswith("cn-") else "com" def _get_handler_of_type(self, type): """ Gets a TemplateHandler type from the registry that can be used to get a string representation of a CloudFormation template. :param type: The type of Template Handler to load :type type: str :return: Instantiated TemplateHandler :rtype: class """ if not self._registry: self._registry = {} for entry_point in iter_entry_points("sceptre.template_handlers", type): self._registry[] = entry_point.load() if type not in self._registry: raise TemplateHandlerNotFoundError('Handler of type "{0}" not found'.format(type)) return self._registry[type]