def to_cloudformation()

in samtranslator/model/stepfunctions/generators.py [0:0]

• 43 lines of code
• 12 McCabe index (conditional complexity)

    def to_cloudformation(self):  # type: ignore[no-untyped-def]
        """
        Constructs and returns the State Machine resource and any additional resources associated with it.

        :returns: a list of resources including the State Machine resource.
        :rtype: list
        """
        resources: List[Any] = [self.state_machine]

        # Defaulting to {} will add the DefinitionSubstitutions field on the transform output even when it is not relevant
        if self.definition_substitutions:
            self.state_machine.DefinitionSubstitutions = self.definition_substitutions

        if self.definition and self.definition_uri:
            raise InvalidResourceException(
                self.logical_id, "Specify either 'Definition' or 'DefinitionUri' property and not both."
            )
        if self.definition:
            processed_definition = deepcopy(self.definition)
            substitutions = self._replace_dynamic_values_with_substitutions(processed_definition)  # type: ignore[no-untyped-call]
            if len(substitutions) > 0:
                if self.state_machine.DefinitionSubstitutions:
                    self.state_machine.DefinitionSubstitutions.update(substitutions)
                else:
                    self.state_machine.DefinitionSubstitutions = substitutions
            self.state_machine.DefinitionString = self._build_definition_string(processed_definition)  # type: ignore[no-untyped-call]
        elif self.definition_uri:
            self.state_machine.DefinitionS3Location = self._construct_definition_uri()
        else:
            raise InvalidResourceException(
                self.logical_id, "Either 'Definition' or 'DefinitionUri' property must be specified."
            )

        if self.role and self.policies:
            raise InvalidResourceException(self.logical_id, self.SFN_INVALID_PROPERTY_BOTH_ROLE_POLICY)
        if self.role:
            self.state_machine.RoleArn = self.role
        else:
            if not self.policies:
                self.policies = []
            execution_role = self._construct_role()
            self.state_machine.RoleArn = execution_role.get_runtime_attr("arn")
            resources.append(execution_role)

        self.state_machine.StateMachineName = self.name
        self.state_machine.StateMachineType = self.type
        self.state_machine.LoggingConfiguration = self.logging
        self.state_machine.TracingConfiguration = self.tracing
        self.state_machine.Tags = self._construct_tag_list()

        managed_traffic_shifting_resources = self._generate_managed_traffic_shifting_resources()
        resources.extend(managed_traffic_shifting_resources)

        event_resources = self._generate_event_resources()
        resources.extend(event_resources)

        return resources