feat: add --express flag to sam delete

samcli/commands/delete/command.py

@@ -65,6 +65,14 @@
     default=None,
     required=False,
 )
+@click.option(
+    "--express/--no-express",
+    default=False,
+    required=False,
+    is_flag=True,
+    help="Use CloudFormation Express mode to speed up stack deletion by completing once resource "
+    "deletion is initiated, without waiting for full cleanup.",
+)
 @aws_creds_options
 @common_options
 @save_params_option
@@ -79,6 +87,7 @@ def cli(
     no_prompts: bool,
     s3_bucket: str,
     s3_prefix: str,
+    express: bool,
     config_env: str,
     config_file: str,
     save_params: bool,
@@ -95,6 +104,7 @@ def cli(
         no_prompts=no_prompts,
         s3_bucket=s3_bucket,
         s3_prefix=s3_prefix,
+        express=express,
     )  # pragma: no cover
 
 
@@ -105,6 +115,7 @@ def do_cli(
     no_prompts: bool,
     s3_bucket: Optional[str],
     s3_prefix: Optional[str],
+    express: bool = False,
 ):
     """
     Implementation of the ``cli`` method
@@ -118,5 +129,6 @@ def do_cli(
         no_prompts=no_prompts,
         s3_bucket=s3_bucket,
         s3_prefix=s3_prefix,
+        express=express,
     ) as delete_context:
         delete_context.run()

samcli/commands/delete/core/options.py

@@ -7,7 +7,7 @@
 from samcli.cli.core.options import ALL_COMMON_OPTIONS, add_common_options_info
 from samcli.cli.row_modifiers import RowDefinition
 
-DELETE_OPTION_NAMES = ["stack_name", "no_prompts", "s3_bucket", "s3_prefix"]
+DELETE_OPTION_NAMES = ["stack_name", "no_prompts", "s3_bucket", "s3_prefix", "express"]
 AWS_CREDENTIAL_OPTION_NAMES = ["region", "profile"]
 CONFIGURATION_OPTION_NAMES = ["config_file", "config_env", "save_params"]
 

samcli/commands/delete/delete_context.py

@@ -37,13 +37,15 @@ def __init__(
         no_prompts: bool,
         s3_bucket: Optional[str],
         s3_prefix: Optional[str],
+        express: bool = False,
     ):
         self.stack_name = stack_name
         self.region = region
         self.profile = profile
         self.no_prompts = no_prompts
         self.s3_bucket = s3_bucket
         self.s3_prefix = s3_prefix
+        self.express = express
         self.cf_utils = None
         self.s3_uploader = None
         self.ecr_uploader = None
@@ -211,15 +213,28 @@ def delete_ecr_companion_stack(self):
             try:
                 # If delete_stack fails and its status changes to DELETE_FAILED, retain
                 # the user input repositories and delete the stack.
-                self.cf_utils.delete_stack(stack_name=self.companion_stack_name)
+                self.cf_utils.delete_stack(
+                    stack_name=self.companion_stack_name, deployment_config=self._deployment_config
+                )
                 self.cf_utils.wait_for_delete(stack_name=self.companion_stack_name)
                 LOG.debug("Deleted ECR Companion Stack: %s", self.companion_stack_name)
 
             except CfDeleteFailedStatusError:
                 LOG.debug("delete_stack resulted failed and so re-try with retain_resources")
-                self.cf_utils.delete_stack(stack_name=self.companion_stack_name, retain_resources=retain_repos)
+                self.cf_utils.delete_stack(
+                    stack_name=self.companion_stack_name,
+                    retain_resources=retain_repos,
+                    deployment_config=self._deployment_config,
+                )
                 self.cf_utils.wait_for_delete(stack_name=self.companion_stack_name)
 
+    @property
+    def _deployment_config(self) -> Optional[dict]:
+        """CloudFormation DeploymentConfig for delete-stack calls, if express mode is enabled."""
+        if self.express:
+            return {"Mode": "EXPRESS"}
+        return None
+
     def delete(self):
         """
         Delete method calls for Cloudformation stacks and S3 and ECR artifacts
@@ -275,13 +290,17 @@ def delete(self):
         # Delete the primary input stack
         try:
             click.echo(f"\t- Deleting Cloudformation stack {self.stack_name}")
-            self.cf_utils.delete_stack(stack_name=self.stack_name)
+            self.cf_utils.delete_stack(stack_name=self.stack_name, deployment_config=self._deployment_config)
             self.cf_utils.wait_for_delete(self.stack_name)
             LOG.debug("Deleted Cloudformation stack: %s", self.stack_name)
 
         except CfDeleteFailedStatusError:
             LOG.debug("delete_stack resulted failed and so re-try with retain_resources")
-            self.cf_utils.delete_stack(stack_name=self.stack_name, retain_resources=retain_resources)
+            self.cf_utils.delete_stack(
+                stack_name=self.stack_name,
+                retain_resources=retain_resources,
+                deployment_config=self._deployment_config,
+            )
             self.cf_utils.wait_for_delete(self.stack_name)
 
         # Warn the user that s3 information is missing and to use --s3 options
@@ -315,6 +334,12 @@ def run(self):
                 LOG.debug("Input stack is deployed, continue deleting")
                 self.delete()
                 click.echo("\nDeleted successfully")
+                if self.express:
+                    click.secho(
+                        "Deleted with CloudFormation Express mode. "
+                        "Some resources may still be removing in the background.",
+                        fg="yellow",
+                    )
             else:
                 LOG.debug("Input stack does not exists on Cloudformation")
                 click.echo(

samcli/lib/delete/cfn_utils.py

@@ -4,7 +4,7 @@
 
 import json
 import logging
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 from botocore.exceptions import BotoCoreError, ClientError, WaiterError
 
@@ -134,7 +134,12 @@ def get_stack_template(self, stack_name: str, stage: str) -> str:
             LOG.error("Unable to get stack details.", exc_info=e)
             raise e
 
-    def delete_stack(self, stack_name: str, retain_resources: Optional[List] = None):
+    def delete_stack(
+        self,
+        stack_name: str,
+        retain_resources: Optional[List] = None,
+        deployment_config: Optional[Dict] = None,
+    ):
         """
         Delete the Cloudformation stack with the given stack_name
 
@@ -144,6 +149,8 @@ def delete_stack(self, stack_name: str, retain_resources: Optional[List] = None)
             str Name or ID of the stack
         retain_resources: Optional[List]
             List of repositories to retain if the stack has DELETE_FAILED status.
+        deployment_config: Optional[Dict]
+            CloudFormation DeploymentConfig, e.g. {"Mode": "EXPRESS"} for Express mode.
 
         Raises
         ------
@@ -152,8 +159,11 @@ def delete_stack(self, stack_name: str, retain_resources: Optional[List] = None)
         """
         if not retain_resources:
             retain_resources = []
+        kwargs: Dict = {"StackName": stack_name, "RetainResources": retain_resources}
+        if deployment_config:
+            kwargs["DeploymentConfig"] = deployment_config
         try:
-            self._client.delete_stack(StackName=stack_name, RetainResources=retain_resources)
+            self._client.delete_stack(**kwargs)
 
         except (ClientError, BotoCoreError) as e:
             # If there are credentials, environment errors,

schema/samcli.json

@@ -1514,7 +1514,7 @@
           "properties": {
             "parameters": {
               "title": "Parameters for the delete command",
-              "description": "Available parameters for the delete command:\n* stack_name:\nThe name of the AWS CloudFormation stack you want to delete.\n* no_prompts:\nSpecify this flag to allow SAM CLI to skip through the guided prompts.\n* s3_bucket:\nThe S3 bucket path you want to delete.\n* s3_prefix:\nThe S3 prefix you want to delete\n* profile:\nSelect a specific profile from your credential file to get AWS credentials.\n* region:\nSet the AWS Region of the service. (e.g. us-east-1)\n* beta_features:\nEnable/Disable beta features.\n* debug:\nTurn on debug logging to print debug message generated by AWS SAM CLI and display timestamps.\n* save_params:\nSave the parameters provided via the command line to the configuration file.",
+              "description": "Available parameters for the delete command:\n* stack_name:\nThe name of the AWS CloudFormation stack you want to delete.\n* no_prompts:\nSpecify this flag to allow SAM CLI to skip through the guided prompts.\n* s3_bucket:\nThe S3 bucket path you want to delete.\n* s3_prefix:\nThe S3 prefix you want to delete\n* express:\nUse CloudFormation Express mode to speed up stack deletion by completing once resource deletion is initiated, without waiting for full cleanup.\n* profile:\nSelect a specific profile from your credential file to get AWS credentials.\n* region:\nSet the AWS Region of the service. (e.g. us-east-1)\n* beta_features:\nEnable/Disable beta features.\n* debug:\nTurn on debug logging to print debug message generated by AWS SAM CLI and display timestamps.\n* save_params:\nSave the parameters provided via the command line to the configuration file.",
               "type": "object",
               "properties": {
                 "stack_name": {
@@ -1537,6 +1537,11 @@
                   "type": "string",
                   "description": "The S3 prefix you want to delete"
                 },
+                "express": {
+                  "title": "express",
+                  "type": "boolean",
+                  "description": "Use CloudFormation Express mode to speed up stack deletion by completing once resource deletion is initiated, without waiting for full cleanup."
+                },
                 "profile": {
                   "title": "profile",
                   "type": "string",

tests/integration/delete/delete_integ_base.py

@@ -20,6 +20,7 @@ def get_delete_command_list(
         no_prompts=None,
         s3_bucket=None,
         s3_prefix=None,
+        express=None,
     ):
         command_list = [get_sam_command(), "delete"]
 
@@ -40,5 +41,9 @@ def get_delete_command_list(
             command_list += ["--s3-bucket", str(s3_bucket)]
         if s3_prefix:
             command_list += ["--s3-prefix", str(s3_prefix)]
+        if express is True:
+            command_list += ["--express"]
+        elif express is False:
+            command_list += ["--no-express"]
 
         return command_list

tests/integration/delete/test_delete_command.py

@@ -554,6 +554,43 @@ def _validate_stack_deleted(self, stack_name: str) -> None:
     #    - Create a stack using guided deploy of type image
     #    - Select no for deleting ECR repository and this will retain the non-empty repository
 
+    @parameterized.expand(["aws-serverless-function.yaml"])
+    def test_delete_express_mode(self, template_file):
+        template_path = self.test_data_path.joinpath(template_file)
+
+        stack_name = self._method_to_stack_name(self.id())
+        self.stacks.append({"name": stack_name})
+
+        deploy_command_list = self.get_deploy_command_list(
+            template_file=template_path,
+            stack_name=stack_name,
+            capabilities="CAPABILITY_IAM",
+            s3_prefix=self.s3_prefix,
+            s3_bucket=self.s3_bucket.name,
+            force_upload=True,
+            notification_arns=self.sns_arn,
+            parameter_overrides="Parameter=Clarity",
+            kms_key_id=self.kms_key,
+            no_execute_changeset=False,
+            tags="integ=true clarity=yes foo_bar=baz",
+            confirm_changeset=False,
+        )
+
+        deploy_process = run_command(deploy_command_list)
+        self.assertEqual(deploy_process.process.returncode, 0)
+
+        delete_command_list = self.get_delete_command_list(
+            stack_name=stack_name,
+            region=self._session.region_name,
+            no_prompts=True,
+            express=True,
+        )
+
+        delete_process_execute = run_command(delete_command_list)
+        self.assertEqual(delete_process_execute.process.returncode, 0)
+        self.assertIn("Deleted with CloudFormation Express mode", delete_process_execute.stdout.decode())
+        self._validate_stack_deleted(stack_name=stack_name)
+
     def _method_to_stack_name(self, method_name):
         """Method expects method name which can be a full path. Eg: test.integration.test_deploy_command.method_name"""
         method_name = method_name.split(".")[-1]

tests/unit/commands/delete/test_command.py

@@ -38,6 +38,36 @@ def test_all_args(self, mock_delete_context, mock_delete_click):
             no_prompts=self.no_prompts,
             s3_bucket=self.s3_bucket,
             s3_prefix=self.s3_prefix,
+            express=False,
+        )
+
+        context_mock.run.assert_called_with()
+        self.assertEqual(context_mock.run.call_count, 1)
+
+    @patch("samcli.commands.delete.command.click")
+    @patch("samcli.commands.delete.delete_context.DeleteContext")
+    def test_express_flag_is_passed_to_context(self, mock_delete_context, mock_delete_click):
+        context_mock = Mock()
+        mock_delete_context.return_value.__enter__.return_value = context_mock
+
+        do_cli(
+            stack_name=self.stack_name,
+            region=self.region,
+            profile=self.profile,
+            no_prompts=self.no_prompts,
+            s3_bucket=self.s3_bucket,
+            s3_prefix=self.s3_prefix,
+            express=True,
+        )
+
+        mock_delete_context.assert_called_with(
+            stack_name=self.stack_name,
+            region=self.region,
+            profile=self.profile,
+            no_prompts=self.no_prompts,
+            s3_bucket=self.s3_bucket,
+            s3_prefix=self.s3_prefix,
+            express=True,
         )
 
         context_mock.run.assert_called_with()

tests/unit/lib/delete/test_cfn_utils.py

@@ -109,6 +109,18 @@ def test_cf_utils_delete_stack_exception(self):
         with self.assertRaises(Exception):
             self.cf_utils.delete_stack("test", ["retain_logical_id"])
 
+    def test_cf_utils_delete_stack_without_deployment_config(self):
+        self.cf_utils._client.delete_stack = MagicMock()
+        self.cf_utils.delete_stack("test")
+        self.cf_utils._client.delete_stack.assert_called_once_with(StackName="test", RetainResources=[])
+
+    def test_cf_utils_delete_stack_with_express_deployment_config(self):
+        self.cf_utils._client.delete_stack = MagicMock()
+        self.cf_utils.delete_stack("test", deployment_config={"Mode": "EXPRESS"})
+        self.cf_utils._client.delete_stack.assert_called_once_with(
+            StackName="test", RetainResources=[], DeploymentConfig={"Mode": "EXPRESS"}
+        )
+
     def test_cf_utils_wait_for_delete_check_waiter_config(self):
         exception = WaiterError(
             name="wait_for_delete",