feat: add --express flag to sam deploy and sam sync for CloudFormation Express mode

CloudFormation Express mode completes stack operations ~4x faster by
returning once resource configuration is applied rather than waiting for
full stabilization. This adds --express/--no-express to both commands:

- sam deploy --express (opt-in, default false)
- sam sync --express (opt-in by default, default true)

When enabled, passes DeploymentConfig={"Mode":"EXPRESS","DisableRollback":...}
to CreateChangeSet/CreateStack/UpdateStack. The DisableRollback value inside
DeploymentConfig is controlled by the existing --disable-rollback flag.

Requires boto3/botocore >= 1.43.38 (PR #9084).

Author: roger-zhangg

samcli/commands/deploy/command.py

@@ -143,6 +143,14 @@
     type=int,
     help="Maximum duration in minutes to wait for the deployment to complete.",
 )
+@click.option(
+    "--express/--no-express",
+    default=False,
+    required=False,
+    is_flag=True,
+    help="Use CloudFormation Express mode to speed up deployments by completing once resource "
+    "configuration is applied, without waiting for full stabilization.",
+)
 @stack_name_option(callback=guided_deploy_stack_name)  # pylint: disable=E1120
 @s3_bucket_option(disable_callback=True)  # pylint: disable=E1120
 @image_repository_option
@@ -204,6 +212,7 @@ def cli(
     disable_rollback,
     on_failure,
     max_wait_duration,
+    express,
 ):
     """
     `sam deploy` command entry point
@@ -241,6 +250,7 @@ def cli(
         disable_rollback,
         on_failure,
         max_wait_duration,
+        express,
     )  # pragma: no cover
 
 
@@ -276,6 +286,7 @@ def do_cli(
     disable_rollback,
     on_failure,
     max_wait_duration,
+    express,
 ):
     """
     Implementation of the ``cli`` method
@@ -387,5 +398,6 @@ def do_cli(
             on_failure=on_failure,
             max_wait_duration=max_wait_duration,
             language_extensions=language_extensions,
+            express=express,
         ) as deploy_context:
             deploy_context.run()

samcli/commands/deploy/core/options.py

@@ -38,6 +38,7 @@
     "on_failure",
     "force_upload",
     "max_wait_duration",
+    "express",
 ]
 
 CONFIGURATION_OPTION_NAMES: List[str] = ["config_env", "config_file"] + SAVE_PARAMS_OPTIONS

samcli/commands/deploy/deploy_context.py

@@ -77,6 +77,7 @@ def __init__(
         on_failure,
         max_wait_duration,
         language_extensions: Optional[bool] = None,
+        express: bool = False,
     ):
         self.template_file = template_file
         self.stack_name = stack_name
@@ -111,6 +112,7 @@ def __init__(
         self._max_template_size = 51200
         self.max_wait_duration = max_wait_duration
         self._language_extensions_enabled = resolve_language_extensions_enabled(language_extensions)
+        self.express = express
 
     def __enter__(self):
         return self
@@ -255,6 +257,11 @@ def deploy(
                 click.secho(f"{resource} has no authentication.", fg="yellow")
 
         assert self.deployer is not None
+        if self.express:
+            deployment_config = {"Mode": "EXPRESS", "DisableRollback": disable_rollback}
+        else:
+            deployment_config = None
+
         if use_changeset:
             try:
                 result, changeset_type = self.deployer.create_and_wait_for_changeset(
@@ -266,6 +273,7 @@ def deploy(
                     notification_arns=notification_arns,
                     s3_uploader=s3_uploader,
                     tags=tags,
+                    deployment_config=deployment_config,
                 )
                 click.echo(self.MSG_SHOWCASE_CHANGESET.format(changeset_id=result["Id"]))
 
@@ -279,7 +287,7 @@ def deploy(
                         return
 
                 marker_time = self.deployer.get_last_event_time(stack_name, 0)
-                self.deployer.execute_changeset(result["Id"], stack_name, disable_rollback)
+                self.deployer.execute_changeset(result["Id"], stack_name, disable_rollback, express=self.express)
                 self.deployer.wait_for_execute(
                     stack_name, changeset_type, disable_rollback, self.on_failure, marker_time, self.max_wait_duration
                 )
@@ -307,6 +315,7 @@ def deploy(
                     s3_uploader=s3_uploader,
                     tags=tags,  # type: ignore[arg-type]
                     on_failure=self.on_failure,
+                    deployment_config=deployment_config,
                 )
                 LOG.debug(result)
 

samcli/commands/sync/command.py

@@ -161,6 +161,14 @@
     help="This option will skip the initial infrastructure deployment if it is not required"
     " by comparing the local template with the template deployed in cloud.",
 )
+@click.option(
+    "--express/--no-express",
+    default=True,
+    required=False,
+    is_flag=True,
+    help="Use CloudFormation Express mode to speed up infrastructure deployments by completing once resource "
+    "configuration is applied, without waiting for full stabilization. Enabled by default.",
+)
 @container_env_var_file_option(cls=ContainerOptions)
 @watch_exclude_option
 @stack_name_option(required=True)  # pylint: disable=E1120
@@ -216,6 +224,7 @@ def cli(
     metadata: dict,
     use_container: bool,
     container_env_var_file: Optional[str],
+    express: bool,
     save_params: bool,
     config_file: str,
     config_env: str,
@@ -262,6 +271,7 @@ def cli(
         build_in_source,
         watch_exclude,
         language_extensions,
+        express,
     )  # pragma: no cover
 
 
@@ -297,6 +307,7 @@ def do_cli(
     build_in_source: Optional[bool],
     watch_exclude: Optional[Dict[str, List[str]]],
     language_extensions: Optional[bool],
+    express: bool = True,
 ) -> None:
     """
     Implementation of the ``cli`` method
@@ -408,6 +419,7 @@ def do_cli(
                     on_failure=None,
                     max_wait_duration=60,
                     language_extensions=language_extensions,
+                    express=express,
                 ) as deploy_context:
                     with SyncContext(
                         dependency_layer,

samcli/lib/deploy/deployer.py

@@ -178,7 +178,16 @@ def has_stack(self, stack_name):
             raise e
 
     def create_changeset(
-        self, stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags
+        self,
+        stack_name,
+        cfn_template,
+        parameter_values,
+        capabilities,
+        role_arn,
+        notification_arns,
+        s3_uploader,
+        tags,
+        deployment_config=None,
     ):
         """
         Call Cloudformation to create a changeset and wait for it to complete
@@ -222,6 +231,9 @@ def create_changeset(
             "Tags": tags,
         }
 
+        if deployment_config:
+            kwargs["DeploymentConfig"] = deployment_config
+
         kwargs = self._process_kwargs(kwargs, s3_uploader, capabilities, role_arn, notification_arns)
         return self._create_change_set(stack_name=stack_name, changeset_type=changeset_type, **kwargs)
 
@@ -369,18 +381,20 @@ def wait_for_changeset(self, changeset_id, stack_name):
 
             raise ChangeSetError(stack_name=stack_name, msg=f"ex: {ex} Status: {status}. Reason: {reason}") from ex
 
-    def execute_changeset(self, changeset_id, stack_name, disable_rollback):
+    def execute_changeset(self, changeset_id, stack_name, disable_rollback, express=False):
         """
         Calls CloudFormation to execute changeset
 
         :param changeset_id: ID of the changeset
         :param stack_name: Name or ID of the stack
         :param disable_rollback: Preserve the state of previously provisioned resources when an operation fails.
+        :param express: If True, skip DisableRollback (managed by DeploymentConfig on the changeset).
         """
         try:
-            self._client.execute_change_set(
-                ChangeSetName=changeset_id, StackName=stack_name, DisableRollback=disable_rollback
-            )
+            kwargs = {"ChangeSetName": changeset_id, "StackName": stack_name}
+            if not express:
+                kwargs["DisableRollback"] = disable_rollback
+            self._client.execute_change_set(**kwargs)
         except botocore.exceptions.ClientError as ex:
             raise DeployFailedError(stack_name=stack_name, msg=str(ex)) from ex
 
@@ -595,11 +609,28 @@ def wait_for_execute(
                 raise ex
 
     def create_and_wait_for_changeset(
-        self, stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags
+        self,
+        stack_name,
+        cfn_template,
+        parameter_values,
+        capabilities,
+        role_arn,
+        notification_arns,
+        s3_uploader,
+        tags,
+        deployment_config=None,
     ):
         try:
             result, changeset_type = self.create_changeset(
-                stack_name, cfn_template, parameter_values, capabilities, role_arn, notification_arns, s3_uploader, tags
+                stack_name,
+                cfn_template,
+                parameter_values,
+                capabilities,
+                role_arn,
+                notification_arns,
+                s3_uploader,
+                tags,
+                deployment_config=deployment_config,
             )
             self.wait_for_changeset(result["Id"], stack_name)
             self.describe_changeset(result["Id"], stack_name)
@@ -646,6 +677,7 @@ def sync(
         s3_uploader: Optional[S3Uploader],
         tags: Optional[Dict],
         on_failure: FailureMode,
+        deployment_config=None,
     ):
         """
         Call the sync command to directly update stack or create stack
@@ -686,6 +718,9 @@ def sync(
             "Tags": tags,
         }
 
+        if deployment_config:
+            kwargs["DeploymentConfig"] = deployment_config
+
         kwargs = self._process_kwargs(kwargs, s3_uploader, capabilities, role_arn, notification_arns)
 
         try:
@@ -695,7 +730,8 @@ def sync(
             msg = ""
 
             if exists:
-                kwargs["DisableRollback"] = disable_rollback  # type: ignore
+                if not deployment_config:
+                    kwargs["DisableRollback"] = disable_rollback  # type: ignore
                 # get the latest stack event, and use 0 in case if the stack does not exist
                 marker_time = self.get_last_event_time(stack_name, 0)
                 result = self.update_stack(**kwargs)
@@ -704,8 +740,8 @@ def sync(
                 )
                 msg = "\nStack update succeeded. Sync infra completed.\n"
             else:
-                # Pass string representation of enum
-                kwargs["OnFailure"] = str(on_failure)
+                if not deployment_config:
+                    kwargs["OnFailure"] = str(on_failure)
 
                 result = self.create_stack(**kwargs)
                 self.wait_for_execute(stack_name, "CREATE", disable_rollback, on_failure=on_failure)