def to_cloudformation()

in samtranslator/model/eventsources/push.py [0:0]

• 35 lines of code
• 17 McCabe index (conditional complexity)

    def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def]
        """If the Api event source has a RestApi property, then simply return the Lambda Permission resource allowing
        API Gateway to call the function. If no RestApi is provided, then additionally inject the path, method, and the
        x-amazon-apigateway-integration into the Swagger body for a provided implicit API.

        :param dict kwargs: a dict containing the implicit RestApi to be modified, should no explicit RestApi \
                be provided.
        :returns: a list of vanilla CloudFormation Resources, to which this Api event expands
        :rtype: list
        """
        resources = []

        function = kwargs.get("function")
        intrinsics_resolver: IntrinsicsResolver = kwargs["intrinsics_resolver"]

        if not function:
            raise TypeError("Missing required keyword argument: function")

        if self.Method is not None:
            # Convert to lower case so that user can specify either GET or get
            self.Method = self.Method.lower()

        resources.extend(self._get_permissions(kwargs))  # type: ignore[no-untyped-call]

        explicit_api = kwargs["explicit_api"]
        api_id = kwargs["api_id"]
        if explicit_api.get("__MANAGE_SWAGGER") or explicit_api.get("MergeDefinitions"):
            self._add_swagger_integration(explicit_api, api_id, function, intrinsics_resolver)  # type: ignore[no-untyped-call]

        swagger_body = explicit_api.get("DefinitionBody")

        # Previously overriding the DefaultAuthorizer in event source Auth would not work properly when DefinitionBody
        # is included in the template. This is because call to update and save the DefinitionBody with any auth
        # overrides was beings skipped due to the check on __MANAGE_SWAGGER above which is only set when no
        # DefinitionBody is set.
        # A new opt-in property, OverrideApiAuth, is added at the event source Auth level which is checked below and
        # makes the necessary call to add_auth_to_swagger() to update and save the DefinitionBody with any auth
        # overrides.
        # We make the call to add_auth_to_swagger() in two separate places because _add_swagger_integration() deals
        # specifically with cases where DefinitionBody is not defined, and below for when DefinitionBody is defined.
        if swagger_body and self.Auth and self.Auth.get("OverrideApiAuth"):
            if not (self.Auth.get("Authorizer") or self.Auth.get("ApiKeyRequired") or self.Auth.get("ResourcePolicy")):
                raise InvalidEventException(
                    self.relative_id,
                    "Must define one of: Authorizer, ApiKeyRequired or ResourcePolicy when using the OverrideApiAuth property.",
                )
            stage = cast(str, self.Stage)
            editor = SwaggerEditor(swagger_body)
            self.add_auth_to_swagger(
                self.Auth,
                explicit_api,
                api_id,
                self.relative_id,
                self.Method,
                self.Path,
                stage,
                editor,
                intrinsics_resolver,
            )
            explicit_api["DefinitionBody"] = editor.swagger
        return resources