def to_cloudformation()

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

• 30 lines of code
• 7 McCabe index (conditional complexity)

    def to_cloudformation(self, **kwargs: Dict[str, Any]) -> List[Resource]:
        """Returns the Scheduler Schedule and an IAM role.

        :param dict kwargs: no existing resources need to be modified
        :returns: a list of vanilla CloudFormation Resources, to which this push event expands
        :rtype: list
        """

        target: Resource

        # For SAM statemachine, the resource object is passed using kwargs["resource"],
        # https://github.com/aws/serverless-application-model/blob/a25933379e1cad3d0df4b35729ee2ec335402fdf/samtranslator/model/stepfunctions/generators.py#L266
        if kwargs.get("resource"):
            target_type = _SchedulerScheduleTargetType.STATE_MACHINE
            target = cast(Resource, kwargs["resource"])
        # for SAM function, the resource object is passed using kwargs["function"],
        # unlike SFN using "resource" keyword argument:
        # https://github.com/aws/serverless-application-model/blob/a25933379e1cad3d0df4b35729ee2ec335402fdf/samtranslator/model/sam_resources.py#L681
        elif kwargs.get("function"):
            target_type = _SchedulerScheduleTargetType.FUNCTION
            target = cast(Resource, kwargs["function"])
        else:
            raise TypeError("Missing required keyword argument: function/resource")

        passthrough_resource_attributes = target.get_passthrough_resource_attributes()

        resources: List[Resource] = []

        scheduler_schedule = self._construct_scheduler_schedule_without_target(passthrough_resource_attributes)
        resources.append(scheduler_schedule)

        dlq_queue_arn: Optional[str] = None
        if self.DeadLetterConfig is not None:
            # The dql config spec is the same as normal "Schedule" event,
            # so continue to use EventBridgeRuleUtils for validation.
            # However, Scheduler doesn't use AWS::SQS::QueuePolicy to grant permissions.
            # so we cannot use EventBridgeRuleUtils.get_dlq_queue_arn_and_resources() here.
            EventBridgeRuleUtils.validate_dlq_config(self.logical_id, self.DeadLetterConfig)  # type: ignore[no-untyped-call]
            dlq_queue_arn, dlq_resources = self._get_dlq_queue_arn_and_resources(
                self.DeadLetterConfig, passthrough_resource_attributes
            )
            resources.extend(dlq_resources)

        execution_role_arn: Union[str, Dict[str, Any]] = self.RoleArn  # type: ignore[assignment]
        if not execution_role_arn:
            execution_role = self._construct_execution_role(
                target, target_type, passthrough_resource_attributes, dlq_queue_arn, self.PermissionsBoundary
            )
            resources.append(execution_role)
            execution_role_arn = execution_role.get_runtime_attr("arn")

        scheduler_schedule.Target = self._construct_scheduler_schedule_target(target, execution_role_arn, dlq_queue_arn)

        return resources