sam_base_provider.py

"""
Base class for SAM Template providers
"""

import logging
from typing import Any, Dict, Iterable, Optional, Union, cast

from samcli.lib.iac.plugins_interfaces import Stack
from samcli.lib.intrinsic_resolver.intrinsic_property_resolver import IntrinsicResolver
from samcli.lib.intrinsic_resolver.intrinsics_symbol_table import IntrinsicsSymbolTable
from samcli.lib.package.ecr_utils import is_ecr_url
from samcli.lib.samlib.resource_metadata_normalizer import ResourceMetadataNormalizer
from samcli.lib.samlib.wrapper import SamTranslatorWrapper
from samcli.lib.utils.resources import (
    AWS_LAMBDA_FUNCTION,
    AWS_LAMBDA_LAYERVERSION,
    AWS_SERVERLESS_FUNCTION,
    AWS_SERVERLESS_LAYERVERSION,
)

LOG = logging.getLogger(__name__)


class SamBaseProvider:
    """
    Base class for SAM Template providers
    """

    DEFAULT_CODEURI = "."

    CODE_PROPERTY_KEYS = {
        AWS_LAMBDA_FUNCTION: "Code",
        AWS_SERVERLESS_FUNCTION: "CodeUri",
        AWS_LAMBDA_LAYERVERSION: "Content",
        AWS_SERVERLESS_LAYERVERSION: "ContentUri",
    }

    IMAGE_PROPERTY_KEYS = {
        AWS_LAMBDA_FUNCTION: "Code",
        AWS_SERVERLESS_FUNCTION: "ImageUri",
    }

    def get(self, name: str) -> Optional[Any]:
        """
        Given name of the function, this method must return the Function object

        :param string name: Name of the function
        :return Function: namedtuple containing the Function information
        """
        raise NotImplementedError("not implemented")

    def get_all(self) -> Iterable:
        """
        Yields all the Lambda functions available in the provider.

        :yields Function: namedtuple containing the function information
        """
        raise NotImplementedError("not implemented")

    @staticmethod
    def _extract_codeuri(resource_properties: Dict, code_property_key: str) -> str:
        """
        Extracts the Function/Layer code path from the Resource Properties

        Parameters
        ----------
        resource_properties dict
            Dictionary representing the Properties of the Resource
        code_property_key str
            Property Key of the code on the Resource

        Returns
        -------
        str
            Representing the local code path
        """
        codeuri = resource_properties.get(code_property_key, SamBaseProvider.DEFAULT_CODEURI)

        if isinstance(codeuri, dict):
            return SamBaseProvider.DEFAULT_CODEURI

        return cast(str, codeuri)

    @staticmethod
    def _is_s3_location(location: Optional[Union[str, Dict]]) -> bool:
        """
        the input could be:
        - CodeUri of Serverless::Function
        - Code of Lambda::Function
        - ContentUri of Serverless::LayerVersion
        - Content of Lambda::LayerVersion
        """
        return (isinstance(location, dict) and ("S3Bucket" in location or "Bucket" in location)) or (
            isinstance(location, str) and location.startswith("s3://")
        )

    @staticmethod
    def _is_ecr_uri(location: Optional[Union[str, Dict]]) -> bool:
        """
        the input could be:
        - ImageUri of Serverless::Function
        - Code of Lambda::Function
        """
        return location is not None and is_ecr_url(
            str(location.get("ImageUri", "")) if isinstance(location, dict) else location
        )

    @staticmethod
    def _warn_code_extraction(resource_type: str, resource_name: str, code_property: str) -> None:
        LOG.warning(
            "The resource %s '%s' has specified S3 location for %s. "
            "It will not be built and SAM CLI does not support invoking it locally.",
            resource_type,
            resource_name,
            code_property,
        )

    @staticmethod
    def _warn_imageuri_extraction(resource_type: str, resource_name: str, image_property: str) -> None:
        LOG.debug(
            "The resource %s '%s' has specified ECR registry image for %s. "
            "It will not be built and SAM CLI does not support invoking it locally.",
            resource_type,
            resource_name,
            image_property,
        )

    @staticmethod
    def _extract_lambda_function_imageuri(resource_properties: Dict, code_property_key: str) -> Optional[str]:
        """
        Extracts the Lambda Function ImageUri from the Resource Properties

        Parameters
        ----------
        resource_properties dict
            Dictionary representing the Properties of the Resource
        code_property_key str
            Property Key of the code on the Resource

        Returns
        -------
        str
            Representing the local imageuri
        """
        return cast(Optional[str], resource_properties.get(code_property_key, dict()).get("ImageUri", None))

    @staticmethod
    def _extract_sam_function_imageuri(resource_properties: Dict[str, str], code_property_key: str) -> Optional[str]:
        """
        Extracts the Serverless Function ImageUri from the Resource Properties

        Parameters
        ----------
        resource_properties dict
            Dictionary representing the Properties of the Resource
        code_property_key str
            Property Key of the code on the Resource

        Returns
        -------
        str
            Representing the local imageuri
        """
        return resource_properties.get(code_property_key)

    @staticmethod
    def get_template(
        template_dict: Dict, parameter_overrides: Optional[Dict[str, str]] = None, use_sam_transform: bool = True
    ) -> Dict:
        """
        Given a SAM template dictionary, return a cleaned copy of the template where SAM plugins have been run
        and parameter values have been substituted.

        Parameters
        ----------
        template_dict : dict
            unprocessed SAM template dictionary

        parameter_overrides: dict
            Optional dictionary of values for template parameters

        use_sam_transform: bool
            Whether to transform the given template with Serverless Application Model. Default is True

        Returns
        -------
        dict
            Processed SAM template
        """
        template_dict = template_dict or {}
        parameters_values = SamBaseProvider._get_parameter_values(template_dict, parameter_overrides)
        if template_dict and use_sam_transform:
            template_dict = SamTranslatorWrapper(template_dict, parameter_values=parameters_values).run_plugins()
        ResourceMetadataNormalizer.normalize(template_dict)

        resolver = IntrinsicResolver(
            template=template_dict,
            symbol_resolver=IntrinsicsSymbolTable(logical_id_translator=parameters_values, template=template_dict),
        )
        template_dict = resolver.resolve_template(ignore_errors=True)
        return template_dict

    @staticmethod
    def get_resolved_template_dict(
        template_dict: Stack,
        parameter_overrides: Optional[Dict[str, str]] = None,
        normalize_resource_metadata: bool = True,
    ) -> Stack:
        """
        Given a SAM template dictionary, return a cleaned copy of the template where SAM plugins have been run
        and parameter values have been substituted.
        Parameters
        ----------
        template_dict : dict
            unprocessed SAM template dictionary
        parameter_overrides: dict
            Optional dictionary of values for template parameters
        normalize_resource_metadata: bool
            flag to normalize resource metadata or not; For package and deploy, we don't need to normalize resource
            metadata, which usually exists in a CDK-synthed template and is used for build and local testing
        Returns
        -------
        dict
            Processed SAM template
            :param template_dict:
            :param parameter_overrides:
            :param normalize_resource_metadata:
        """
        template_dict = template_dict or Stack()
        parameters_values = SamBaseProvider._get_parameter_values(template_dict, parameter_overrides)
        if template_dict:
            template_dict = SamTranslatorWrapper(template_dict, parameter_values=parameters_values).run_plugins()
        if normalize_resource_metadata:
            ResourceMetadataNormalizer.normalize(template_dict)

        resolver = IntrinsicResolver(
            template=template_dict,
            symbol_resolver=IntrinsicsSymbolTable(logical_id_translator=parameters_values, template=template_dict),
        )
        template_dict = resolver.resolve_template(ignore_errors=True)
        return template_dict

    @staticmethod
    def _get_parameter_values(template_dict: Any, parameter_overrides: Optional[Dict]) -> Dict:
        """
        Construct a final list of values for CloudFormation template parameters based on user-supplied values,
        default values provided in template, and sane defaults for pseudo-parameters.

        Parameters
        ----------
        template_dict : dict
            SAM template dictionary

        parameter_overrides : dict
            User-supplied values for CloudFormation template parameters

        Returns
        -------
        dict
            Values for template parameters to substitute in template with
        """

        default_values = SamBaseProvider._get_default_parameter_values(template_dict)

        # NOTE: Ordering of following statements is important. It makes sure that any user-supplied values
        # override the defaults
        parameter_values = {}
        parameter_values.update(IntrinsicsSymbolTable.DEFAULT_PSEUDO_PARAM_VALUES)
        parameter_values.update(default_values)
        parameter_values.update(parameter_overrides or {})

        return parameter_values

    @staticmethod
    def _get_default_parameter_values(sam_template: Dict) -> Dict:
        """
        Method to read default values for template parameters and return it
        Example:
        If the template contains the following parameters defined
        Parameters:
            Param1:
                Type: String
                Default: default_value1
            Param2:
                Type: String
                Default: default_value2

        then, this method will grab default value for Param1 and return the following result:
        {
            Param1: "default_value1",
            Param2: "default_value2"
        }
        :param dict sam_template: SAM template
        :return dict: Default values for parameters
        """

        default_values: Dict = {}

        parameter_definition = sam_template.get("Parameters", None)
        if not parameter_definition or not isinstance(parameter_definition, dict):
            LOG.debug("No Parameters detected in the template")
            return default_values

        for param_name, value in parameter_definition.items():
            if isinstance(value, dict) and "Default" in value:
                default_values[param_name] = value["Default"]

        LOG.debug("Collected default values for parameters: %s", default_values)
        return default_values