Source code for sceptre.template_handlers.http

# -*- coding: utf-8 -*-
import pathlib
import tempfile
from urllib.parse import urlparse

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

import sceptre.template_handlers.helper as helper
from sceptre.exceptions import UnsupportedTemplateFileTypeError
from sceptre.template_handlers import TemplateHandler

HANDLER_OPTION_KEY = "http_template_handler"

[docs]class Http(TemplateHandler): """ Template handler that can resolve templates from the web. Standard CFN templates with extension (.json, .yaml, .template) are deployed directly from memory while references to jinja (.j2) and python (.py) templates are downloaded, transformed into CFN templates then deployed to AWS. """ def __init__(self, *args, **kwargs): super(Http, self).__init__(*args, **kwargs)
[docs] def schema(self): return { "type": "object", "properties": {"url": {"type": "string"}}, "required": ["url"], }
[docs] def handle(self): """ handle template from web """ url = self.arguments["url"] path = pathlib.Path(urlparse(url).path) if path.suffix not in self.supported_template_extensions: raise UnsupportedTemplateFileTypeError( "Template has file extension %s. Only %s are supported.", path.suffix, ",".join(self.supported_template_extensions), ) retries = self._get_handler_option( HANDLER_RETRIES_OPTION_PARAM, DEFAULT_RETRIES_OPTION ) timeout = self._get_handler_option( HANDLER_TIMEOUT_OPTION_PARAM, DEFAULT_TIMEOUT_OPTION ) try: template = self._get_template(url, retries=retries, timeout=timeout) if ( path.suffix in self.jinja_template_extensions + self.python_template_extensions ): file = tempfile.NamedTemporaryFile(prefix=path.stem) self.logger.debug("Template file saved to: %s", with file as f: f.write(template) if path.suffix in self.jinja_template_extensions: template = helper.render_jinja_template(, {"sceptre_user_data": self.sceptre_user_data}, self.stack_group_config.get("j2_environment", {}), ) elif path.suffix in self.python_template_extensions: template = helper.call_sceptre_handler(, self.sceptre_user_data ) except Exception as e: helper.print_template_traceback(path) raise e return template
def _get_template(self, url: str, retries: int, timeout: int) -> str: """ Get template from the web :param url: The url to the template :param retries: The number of retry attempts. :param timeout: The timeout for the session in seconds. :raises: :class:`requests.exceptions.HTTPError`: When a download error occurs """ self.logger.debug("Downloading file from: %s", url) session = self._get_retry_session(retries=retries) response = session.get(url, timeout=timeout) # If the response was unsuccessful, raise an error. response.raise_for_status() return response.content def _get_retry_session( self, retries, backoff_factor=0.3, status_forcelist=(429, 500, 502, 503, 504), session=None, ): """ Get a request session with retries. Retry options are explained in the request libraries """ session = session or requests.Session() retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=status_forcelist, ) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) return session def _get_handler_option(self, name, default): """ Get the template handler options :param url: The option name :type: str :param default: The default value if option is not set. :rtype: int """ if HANDLER_OPTION_KEY in self.stack_group_config: option = self.stack_group_config.get(HANDLER_OPTION_KEY) if name in option: return option.get(name) return default