def _create_or_update_stack()

in src/rpdk/core/upload.py [0:0]

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

    def _create_or_update_stack(self, template, stack_name):
        args = {"StackName": stack_name, "TemplateBody": template}
        # attempt to create stack. if the stack already exists, try to update it
        LOG.info("Creating %s", stack_name)
        try:
            result = self.cfn_client.create_stack(
                **args,
                EnableTerminationProtection=True,
                Capabilities=["CAPABILITY_IAM"],
            )
        except self.cfn_client.exceptions.AlreadyExistsException:
            LOG.info("%s already exists. " "Attempting to update", stack_name)
            try:
                result = self.cfn_client.update_stack(
                    **args, Capabilities=["CAPABILITY_IAM"]
                )
            except ClientError as e:
                # if the update is a noop, don't do anything else
                msg = str(e)
                if "No updates are to be performed" in msg:
                    LOG.info("%s stack is up to date", stack_name)
                    stack_id = stack_name
                else:
                    LOG.debug(
                        "%s stack update " "resulted in unknown ClientError",
                        stack_name,
                        exc_info=e,
                    )
                    raise DownstreamError("Unknown CloudFormation error") from e
            else:
                stack_id = result["StackId"]
                self._wait_for_stack(
                    stack_id,
                    "stack_update_complete",
                    "{} stack is up to date".format(stack_name),
                )
        except ClientError as e:
            LOG.debug(
                "%s stack create " "resulted in unknown ClientError",
                stack_name,
                exc_info=e,
            )
            raise DownstreamError("Unknown CloudFormation error") from e
        else:
            stack_id = result["StackId"]
            self._wait_for_stack(
                stack_id,
                "stack_create_complete",
                "{} stack was successfully created".format(stack_name),
            )

        return stack_id