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
import sys

import sceptre.helpers

from sceptre.exceptions import TemplateHandlerNotFoundError
from sceptre.logging import StackLoggerAdapter

[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 = StackLoggerAdapter(logging.getLogger(__name__), 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.helpers.gen_repr( self, class_label="sceptre.template.Template", attributes=["name", "handler_config", "sceptre_user_data", "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 bucket_name is set to None, it should be ignored and not uploaded. if self.s3_details and self.s3_details.get("bucket_name"): 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" @staticmethod def _iterate_entry_points(group, name): """ Helper to determine whether to use pkg_resources or importlib.metadata. """ if sys.version_info < (3, 10): from pkg_resources import iter_entry_points return iter_entry_points(group, name) else: from importlib.metadata import entry_points return entry_points(group=group, name=name) 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 self._iterate_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]