def on_before_transform_template()

in samtranslator/plugins/application/serverless_app_plugin.py [0:0]

• 37 lines of code
• 11 McCabe index (conditional complexity)

    def on_before_transform_template(self, template_dict):  # type: ignore[no-untyped-def]
        """
        Hook method that gets called before the SAM template is processed.
        The template has passed the validation and is guaranteed to contain a non-empty "Resources" section.

        This plugin needs to run as soon as possible to allow some time for templates to become available.
        This verifies that the user has access to all specified applications.

        :param dict template_dict: Dictionary of the SAM template
        """
        template = SamTemplate(template_dict)
        intrinsic_resolvers = self._get_intrinsic_resolvers(template_dict.get("Mappings", {}))  # type: ignore[no-untyped-call]

        service_call = None
        service_call = (
            self._handle_get_application_request if self._validate_only else self._handle_create_cfn_template_request
        )
        for logical_id, app in template.iterate({SamResourceType.Application.value}):
            if not self._can_process_application(app):  # type: ignore[no-untyped-call]
                # Handle these cases in the on_before_transform_resource event
                continue

            app_id = self._replace_value(  # type: ignore[no-untyped-call]
                app.properties[self.LOCATION_KEY], self.APPLICATION_ID_KEY, intrinsic_resolvers
            )

            semver = self._replace_value(  # type: ignore[no-untyped-call]
                app.properties[self.LOCATION_KEY], self.SEMANTIC_VERSION_KEY, intrinsic_resolvers
            )

            key = self._make_app_key(app_id, semver)

            if isinstance(app_id, dict) or isinstance(semver, dict):
                self._applications[key] = False
                continue

            if key not in self._applications:
                try:
                    # Examine the type of ApplicationId and SemanticVersion
                    # before calling SAR API.
                    sam_expect(app_id, logical_id, "Location.ApplicationId").to_be_a_string()
                    sam_expect(semver, logical_id, "Location.SemanticVersion").to_be_a_string()
                    if not RegionConfiguration.is_service_supported("serverlessrepo"):  # type: ignore[no-untyped-call]
                        raise InvalidResourceException(
                            logical_id, "Serverless Application Repository is not available in this region."
                        )
                    # SSM Pattern found here https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/dynamic-references.html
                    ssm_pattern = r"{{resolve:ssm:[a-zA-Z0-9_.\-/]+(:\d+)?}}"
                    if re.search(ssm_pattern, app_id):
                        raise InvalidResourceException(
                            logical_id,
                            "Serverless Application Repostiory does not support dynamic reference in 'ApplicationId' property.",
                        )

                    self._make_service_call_with_retry(service_call, app_id, semver, key, logical_id)  # type: ignore[no-untyped-call]
                except InvalidResourceException as e:
                    # Catch all InvalidResourceExceptions, raise those in the before_resource_transform target.
                    self._applications[key] = e