[containerapp] az containerapp [create/update] --environment-mode WorkloadProfiles Add environment mode to az containerapp

src/containerapp/HISTORY.rst

@@ -4,6 +4,7 @@ Release History
 ===============
 upcoming
 ++++++
+* 'az containerapp env --environment-mode': Add environment mode to create and update commands
 
 1.3.0b2
 ++++++

src/containerapp/azext_containerapp/_help.py

@@ -331,14 +331,14 @@
               --logs-workspace-id myLogsWorkspaceID \\
               --logs-workspace-key myLogsWorkspaceKey \\
               --location eastus2
-    - name: Create an environment with workload profiles enabled.
+    - name: Create an environment with workload profiles enabled (default mode).
       text: |
           az containerapp env create -n MyContainerappEnvironment -g MyResourceGroup \\
-              --location eastus2 --enable-workload-profiles
-    - name: Create an environment without workload profiles enabled.
+              --location eastus2 --environment-mode WorkloadProfiles
+    - name: Create an environment in consumption-only mode (no workload profiles).
       text: |
           az containerapp env create -n MyContainerappEnvironment -g MyResourceGroup \\
-              --location eastus2 --enable-workload-profiles false
+              --location eastus2 --environment-mode ConsumptionOnly
     - name: Create an environment with system assigned and user assigned identity.
       text: |
           az containerapp env create -n MyContainerappEnvironment -g MyResourceGroup \\

src/containerapp/azext_containerapp/_params.py

@@ -174,6 +174,7 @@ def load_arguments(self, _):
     with self.argument_context('containerapp env') as c:
         c.argument('public_network_access', arg_type=get_enum_type(['Enabled', 'Disabled']),
                    help="Allow or block all public traffic", is_preview=True)
+        c.argument('environment_mode', options_list=["--environment-mode"], help="Mode of the environment.", is_preview=True)
 
     with self.argument_context('containerapp env', arg_group='Custom Domain') as c:
         c.argument('certificate_identity', options_list=['--custom-domain-certificate-identity', '--certificate-identity'],

src/containerapp/azext_containerapp/_up_utils.py

@@ -183,6 +183,7 @@ def __init__(
         workload_profile_type=None,
         workload_profile_name=None,
         is_env_for_azml_app=None,
+        environment_mode=None
     ):
         self.resource_type = None
         super().__init__(cmd, name, resource_group, exists)
@@ -211,6 +212,7 @@ def __init__(
         self.workload_profile_type = workload_profile_type
         self.workload_profile_name = workload_profile_name
         self.is_env_for_azml_app = is_env_for_azml_app
+        self.environment_mode = environment_mode
 
     def set_name(self, name_or_rid):
         if is_valid_resource_id(name_or_rid):
@@ -283,6 +285,7 @@ def create(self):  # pylint: disable=arguments-differ
                 workload_profile_type=self.workload_profile_type,
                 workload_profile_name=self.workload_profile_name,
                 is_env_for_azml_app=self.is_env_for_azml_app,
+                environment_mode=self.environment_mode
             )
             self.exists = True
 

src/containerapp/azext_containerapp/_utils.py

@@ -512,6 +512,19 @@ def get_cluster_extension(cmd, cluster_extension_id=None):
         extension_name=resource_name)
 
 
+def validate_environment_mode_and_workload_profiles_compatible(environment_mode, workload_profiles_enabled):
+    # If only environment_mode is specified, derive enable_workload_profiles from it
+    if environment_mode is not None:
+        is_environment_mode_workload_profiles_enabled = environment_mode.lower() != 'consumptiononly'
+
+        # Check for conflicts when both are specified
+        if workload_profiles_enabled is not None:
+            if not is_environment_mode_workload_profiles_enabled and workload_profiles_enabled:
+                raise ValidationError("Cannot use '--enable-workload-profiles' with '--environment-mode ConsumptionOnly'. Please use '--environment-mode' alone.")
+            if is_environment_mode_workload_profiles_enabled and not workload_profiles_enabled:
+                raise ValidationError("Cannot use '--enable-workload-profiles false' with '--environment-mode {}'. Please use '--environment-mode' alone.".format(environment_mode))
+
+
 def validate_custom_location(cmd, custom_location=None):
     if not is_valid_resource_id(custom_location):
         raise ValidationError('{} is not a valid Azure resource ID.'.format(custom_location))

src/containerapp/azext_containerapp/containerapp_env_decorator.py

@@ -13,7 +13,7 @@
 from azure.cli.core.commands.client_factory import get_subscription_id
 
 from ._models import ManagedServiceIdentity, CustomDomainConfiguration
-from ._utils import safe_get
+from ._utils import safe_get, validate_environment_mode_and_workload_profiles_compatible
 from ._client_factory import handle_non_404_status_code_exception
 
 logger = get_logger(__name__)
@@ -31,7 +31,7 @@ def construct_payload(self):
         self.managed_env_def["tags"] = self.get_argument_tags()
         self.managed_env_def["properties"]["zoneRedundant"] = self.get_argument_zone_redundant()
 
-        self.set_up_workload_profiles()
+        self._set_up_workload_profiles_and_environment_mode()
 
         if self.get_argument_instrumentation_key() is not None:
             self.managed_env_def["properties"]["daprAIInstrumentationKey"] = self.get_argument_instrumentation_key()
@@ -43,24 +43,37 @@ def construct_payload(self):
         # copy end
 
         # overwrite custom_domain_configuration
-        self.set_up_custom_domain_configuration()
+        self._set_up_custom_domain_configuration()
 
-        self.set_up_infrastructure_resource_group()
-        self.set_up_dynamic_json_columns()
-        self.set_up_managed_identity()
-        self.set_up_public_network_access()
+        self._set_up_infrastructure_resource_group()
+        self._set_up_dynamic_json_columns()
+        self._set_up_managed_identity()
+        self._set_up_public_network_access()
 
     def validate_arguments(self):
         super().validate_arguments()
 
+        # Check if user explicitly provided --enable-workload-profiles
+        safe_params = self.cmd.cli_ctx.data.get('safe_params', [])
+        user_provided_workload_profiles = '-w' in safe_params or '--enable-workload-profiles' in safe_params
+
+        # Only pass enable_workload_profiles if user explicitly provided it
+        workload_profiles_value = self.get_argument_enable_workload_profiles() if user_provided_workload_profiles else None
+
+        # Resolve environment_mode and enable_workload_profiles
+        validate_environment_mode_and_workload_profiles_compatible(
+            self.get_argument_environment_mode(),
+            workload_profiles_value
+        )
+
         # Infrastructure Resource Group
         if self.get_argument_infrastructure_resource_group() is not None:
             if not self.get_argument_infrastructure_subnet_resource_id():
                 raise RequiredArgumentMissingError("Cannot use --infrastructure-resource-group/-i without "
                                                    "--infrastructure-subnet-resource-id/-s")
-            if not self.get_argument_enable_workload_profiles():
-                raise RequiredArgumentMissingError("Cannot use --infrastructure-resource-group/-i without "
-                                                   "--enable-workload-profiles/-w")
+            if not self._get_effective_workload_profiles():
+                raise RequiredArgumentMissingError("Cannot use --infrastructure-resource-group/-i with "
+                                                   "--environment-mode ConsumptionOnly")
 
         # validate custom domain configuration
         if self.get_argument_hostname():
@@ -69,20 +82,21 @@ def validate_arguments(self):
             if (not self.get_argument_certificate_file()) and (not self.get_argument_certificate_key_vault_url()):
                 raise ValidationError("Either --certificate-file or --certificate-akv-url should be set when --dns-suffix is set")
 
-    def set_up_public_network_access(self):
+    def _set_up_public_network_access(self):
         if self.get_argument_public_network_access():
             safe_set(self.managed_env_def, "properties", "publicNetworkAccess",
                      value=self.get_argument_public_network_access())
 
-    def set_up_dynamic_json_columns(self):
+    def _set_up_dynamic_json_columns(self):
         if self.get_argument_logs_destination() == "log-analytics" and self.get_argument_logs_dynamic_json_columns() is not None:
             safe_set(self.managed_env_def, "properties", "appLogsConfiguration", "logAnalyticsConfiguration", "dynamicJsonColumns", value=self.get_argument_logs_dynamic_json_columns())
 
-    def set_up_infrastructure_resource_group(self):
-        if self.get_argument_enable_workload_profiles() and self.get_argument_infrastructure_subnet_resource_id() is not None:
+    def _set_up_infrastructure_resource_group(self):
+        effective_workload_profiles = self._get_effective_workload_profiles()
+        if effective_workload_profiles and self.get_argument_infrastructure_subnet_resource_id() is not None:
             self.managed_env_def["properties"]["infrastructureResourceGroup"] = self.get_argument_infrastructure_resource_group()
 
-    def set_up_managed_identity(self):
+    def _set_up_managed_identity(self):
         if self.get_argument_system_assigned() or self.get_argument_user_assigned():
             identity_def = ManagedServiceIdentity
             identity_def["type"] = "None"
@@ -109,19 +123,25 @@ def set_up_managed_identity(self):
                     identity_def["userAssignedIdentities"][r] = {}  # pylint: disable=unsupported-assignment-operation
             self.managed_env_def["identity"] = identity_def
 
-    def set_up_workload_profiles(self):
-        if self.get_argument_enable_workload_profiles():
-            # If the environment exists, infer the environment type
-            existing_environment = None
-            try:
-                existing_environment = self.client.show(cmd=self.cmd, resource_group_name=self.get_argument_resource_group_name(), name=self.get_argument_name())
-            except Exception as e:
-                handle_non_404_status_code_exception(e)
-
-            if existing_environment and safe_get(existing_environment, "properties", "workloadProfiles") is None:
-                # check if input params include -w/--enable-workload-profiles
-                if self.cmd.cli_ctx.data.get('safe_params') and ('-w' in self.cmd.cli_ctx.data.get('safe_params') or '--enable-workload-profiles' in self.cmd.cli_ctx.data.get('safe_params')):
-                    raise ValidationError(f"Existing environment {self.get_argument_name()} cannot enable workload profiles. If you want to use Consumption and Dedicated environment, please create a new one.")
+    # environment mode and workload profiles are coupled, so set them up together
+    def _set_up_workload_profiles_and_environment_mode(self):
+        # Use resolved effective value (supports both --environment-mode and --enable-workload-profiles)
+        effective_workload_profiles = self._get_effective_workload_profiles()
+        # If the environment exists, infer the environment type
+        existing_environment = None
+        environment_mode = self.get_argument_environment_mode()
+        try:
+            existing_environment = self.client.show(cmd=self.cmd, resource_group_name=self.get_argument_resource_group_name(), name=self.get_argument_name())
+        except Exception as e:
+            handle_non_404_status_code_exception(e)
+
+        if effective_workload_profiles:
+            # Check if existing environment is ConsumptionOnly (no workload profiles)
+            if existing_environment:
+                if safe_get(existing_environment, "properties", "workloadProfiles") is None:
+                    if self.cmd.cli_ctx.data.get('safe_params') and ('-w' in self.cmd.cli_ctx.data.get('safe_params') or '--enable-workload-profiles' in self.cmd.cli_ctx.data.get('safe_params')):
+                        # User is trying to enable workload profiles on a ConsumptionOnly environment
+                        raise ValidationError(f"Existing environment {self.get_argument_name()} cannot enable workload profiles. If you want to use Consumption and Dedicated environment, please create a new one.")
                 return
 
             workload_profiles = get_default_workload_profiles(self.cmd, self.get_argument_location())
@@ -148,8 +168,18 @@ def set_up_workload_profiles(self):
                     }
                     workload_profiles.append(serverless_gpu_profile)
             self.managed_env_def["properties"]["workloadProfiles"] = workload_profiles
+        else:
+            # Check if existing environment is WorkloadProfiles
+            if existing_environment:
+                if safe_get(existing_environment, "properties", "workloadProfiles") is not None:
+                    # User is trying to enable workload profiles on a ConsumptionOnly environment
+                    raise ValidationError(f"Existing environment {self.get_argument_name()} cannot be a Consumption only environment. If you want to use Consumption only environment, please create a new one.")
+                return
 
-    def set_up_custom_domain_configuration(self):
+        if environment_mode:
+            self.managed_env_def["properties"]["environmentMode"] = environment_mode
+
+    def _set_up_custom_domain_configuration(self):
         if self.get_argument_hostname():
             custom_domain = CustomDomainConfiguration
             custom_domain["dnsSuffix"] = self.get_argument_hostname()
@@ -175,6 +205,9 @@ def set_up_custom_domain_configuration(self):
     def get_argument_enable_workload_profiles(self):
         return self.get_param("enable_workload_profiles")
 
+    def get_argument_environment_mode(self):
+        return self.get_param("environment_mode")
+
     def get_argument_enable_dedicated_gpu(self):
         return self.get_param("enable_dedicated_gpu")
 
@@ -205,6 +238,26 @@ def get_argument_workload_profile_type(self):
     def get_argument_workload_profile_name(self):
         return self.get_param("workload_profile_name")
 
+    def _get_effective_workload_profiles(self):
+
+        safe_params = self.cmd.cli_ctx.data.get('safe_params', [])
+
+        # First check if user provided --environment-mode
+        if '--environment-mode' in safe_params:
+            environment_mode = self.get_argument_environment_mode()
+            if environment_mode:
+                # WorkloadProfiles mode = workload profiles enabled
+                # ConsumptionOnly = workload profiles disabled
+                return environment_mode.lower() == "workloadprofiles"
+
+        # Fallback: check if user explicitly provided --enable-workload-profiles
+        user_provided_wp = '-w' in safe_params or '--enable-workload-profiles' in safe_params
+        if user_provided_wp:
+            return self.get_argument_enable_workload_profiles()
+
+        # Default to True if neither --environment-mode nor --enable-workload-profiles was provided
+        return True
+
 
 class ContainerappEnvPreviewUpdateDecorator(ContainerAppEnvUpdateDecorator):
     def validate_arguments(self):
@@ -218,6 +271,12 @@ def construct_payload(self):
         super().construct_payload()
 
         self.set_up_public_network_access()
+        self._set_up_environment_mode()
+
+    def _set_up_environment_mode(self):
+        environment_mode = self.get_argument_environment_mode()
+        if environment_mode:
+            safe_set(self.managed_env_def, "properties", "environmentMode", value=environment_mode)
 
     def set_up_public_network_access(self):
         if self.get_argument_public_network_access():
@@ -275,3 +334,6 @@ def get_argument_certificate_key_vault_url(self):
 
     def get_argument_public_network_access(self):
         return self.get_param("public_network_access")
+
+    def get_argument_environment_mode(self):
+        return self.get_param("environment_mode")

src/containerapp/azext_containerapp/custom.py

@@ -763,7 +763,8 @@ def create_managed_environment(cmd,
                                logs_dynamic_json_columns=False,
                                system_assigned=False,
                                user_assigned=None,
-                               public_network_access=None):
+                               public_network_access=None,
+                               environment_mode=None):
     return create_managed_environment_logic(
         cmd=cmd,
         name=name,
@@ -797,7 +798,8 @@ def create_managed_environment(cmd,
         logs_dynamic_json_columns=logs_dynamic_json_columns,
         system_assigned=system_assigned,
         user_assigned=user_assigned,
-        public_network_access=public_network_access
+        public_network_access=public_network_access,
+        environment_mode=environment_mode
     )
 
 
@@ -836,7 +838,8 @@ def create_managed_environment_logic(cmd,
                                      public_network_access=None,
                                      workload_profile_type=None,
                                      workload_profile_name=None,
-                                     is_env_for_azml_app=False):
+                                     is_env_for_azml_app=False,
+                                     environment_mode=None):
     raw_parameters = locals()
     containerapp_env_create_decorator = ContainerappEnvPreviewCreateDecorator(
         cmd=cmd,
@@ -875,7 +878,8 @@ def update_managed_environment(cmd,
                                p2p_encryption_enabled=None,
                                no_wait=False,
                                logs_dynamic_json_columns=None,
-                               public_network_access=None):
+                               public_network_access=None,
+                               environment_mode=None):
     raw_parameters = locals()
     containerapp_env_update_decorator = ContainerappEnvPreviewUpdateDecorator(
         cmd=cmd,