[SQL] Add short-term backup retention immutability support

src/azure-cli/azure/cli/command_modules/sql/_help.py

@@ -214,17 +214,23 @@
 helps['sql db str-policy set'] = """
 type: command
 short-summary: Update short term retention settings for a live database.
+long-summary: Locking backup immutability is irreversible. Confirmation is required only when --lock-immutability is true. Use --yes to skip confirmation. Specify at least one retention setting or --lock-immutability.
 examples:
   - name: Set short term retention for a live database.
-    text: az sql db str-policy set -g mygroup -s myserver -n mydb --retention-days retentionindays --diffbackup-hours diffbackuphours
+    text: az sql db str-policy set --resource-group mygroup --server myserver --name mydb --retention-days 7 --diffbackup-hours 24
+  - name: Lock backup immutability without changing the retention settings, skipping confirmation.
+    text: az sql db str-policy set --resource-group mygroup --server myserver --name mydb --lock-immutability true --yes
 """
 
 helps['sql db str-policy show'] = """
 type: command
 short-summary: Show the short term retention policy for a live database.
+long-summary: The immutabilityStatus output property reports Disabled, Enabled or Locked when returned by the service.
 examples:
   - name: Show short term retention policy for a live database.
     text: az sql db str-policy show -g mygroup -s myserver -n mydb
+  - name: Show the current backup immutability status.
+    text: az sql db str-policy show --resource-group mygroup --server myserver --name mydb --query immutabilityStatus
 """
 
 helps['sql db str-policy wait'] = """
@@ -1212,16 +1218,20 @@
 helps['sql midb short-term-retention-policy set'] = """
 type: command
 short-summary: Update short term retention for automated backups on a single database.
+long-summary: Locking backup immutability is irreversible and is supported only for live managed databases. Confirmation is required only when --lock-immutability is true; use --yes to skip it. Do not specify --lock-immutability with --deleted-time, even when false. Retention cannot be reduced on a dropped database whose backup immutability is locked.
 examples:
   - name: Set backup short term retention for live managed database.
-    text: az sql midb short-term-retention-policy set -g mygroup --mi myinstance -n mymanageddb --retention-days retentionindays
+    text: az sql midb short-term-retention-policy set -g mygroup --mi myinstance -n mymanageddb --retention-days 7
   - name: Set backup short term retention for dropped managed database.
-    text: az sql midb short-term-retention-policy set -g mygroup --mi myinstance -n mymanageddb --deleted-time "2018-05-20T05:34:22" --retention-days retentionindays
+    text: az sql midb short-term-retention-policy set -g mygroup --mi myinstance -n mymanageddb --deleted-time "2018-05-20T05:34:22" --retention-days 7
+  - name: Lock backup immutability for a live managed database, skipping confirmation.
+    text: az sql midb short-term-retention-policy set --resource-group mygroup --mi myinstance --name mymanageddb --retention-days 7 --lock-immutability true --yes
 """
 
 helps['sql midb short-term-retention-policy show'] = """
 type: command
 short-summary: Show short term retention for automated backups on a single database.
+long-summary: The immutabilityStatus output property reports Disabled, Enabled or Locked for live and restorable dropped managed databases when returned by the service.
 examples:
   - name: Shows backup short term retention for live managed database.
     text: az sql midb short-term-retention-policy show -g mygroup --mi myinstance -n mymanageddb

src/azure-cli/azure/cli/command_modules/sql/_params.py

@@ -1506,24 +1506,29 @@ def _configure_security_policy_storage_params(arg_ctx):
     #                sql db str                   #
     ###############################################
     with self.argument_context('sql db str-policy set') as c:
-        create_args_for_complex_type(
-            c, 'parameters', Database, [
-                'retention_days',
-                'diffbackup_hours'
-            ])
-
         c.argument(
             'retention_days',
             options_list=['--retention-days'],
-            required=True,
+            type=int,
+            required=False,
             help='New backup short term retention policy retention in days.'
             'Valid retention days for live database of (DTU) Basic can be 1-7 days; Rest models can be 1-35 days.')
-
         c.argument(
             'diffbackup_hours',
             options_list=['--diffbackup-hours'],
+            type=int,
             help='New backup short term retention policy differential backup interval in hours.'
             'Valid differential backup interval for live database can be 12 or 24 hours.')
+        c.argument(
+            'lock_immutability',
+            arg_type=get_three_state_flag(),
+            help='Whether to lock backup immutability. Locking is irreversible. '
+                 'Setting false does not unlock an already locked policy.')
+        c.argument(
+            'yes',
+            options_list=['--yes', '-y'],
+            action='store_true',
+            help='Do not prompt for confirmation when locking backup immutability.')
 
     ###############################################
     #                sql dw                       #
@@ -2825,24 +2830,28 @@ class ContainerIdentityType(Enum):
             help='The id of recoverable database from geo-replicated instance')
 
     with self.argument_context('sql midb short-term-retention-policy set') as c:
-        create_args_for_complex_type(
-            c, 'parameters', ManagedDatabase, [
-                'deleted_time',
-                'retention_days'
-            ])
-
         c.argument(
             'deleted_time',
             options_list=['--deleted-time'],
-            help='If specified, updates retention days for a deleted database, instead of an existing database.'
-            'Must match the deleted time of a deleted database on the source Managed Instance.')
-
+            help='If specified, updates retention days for a deleted database, instead of an existing database. '
+                 'Must match the deleted time of a deleted database on the source Managed Instance.')
         c.argument(
             'retention_days',
             options_list=['--retention-days'],
+            type=int,
             required=True,
-            help='New backup short term retention policy in days.'
-            'Valid policy for live database is 7-35 days, valid policy for dropped databases is 0-35 days.')
+            help='New backup short term retention policy in days. '
+                 'Valid policy for live database is 7-35 days, valid policy for dropped databases is 0-35 days.')
+        c.argument(
+            'lock_immutability',
+            arg_type=get_three_state_flag(),
+            help='Whether to lock backup immutability for a live managed database. '
+                 'Locking is irreversible. Cannot be used with --deleted-time, even when false.')
+        c.argument(
+            'yes',
+            options_list=['--yes', '-y'],
+            action='store_true',
+            help='Do not prompt for confirmation when locking backup immutability.')
 
     with self.argument_context('sql midb short-term-retention-policy show') as c:
         c.argument(

src/azure-cli/azure/cli/command_modules/sql/_str_policy.py

@@ -0,0 +1,157 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+"""SDK-independent STR operations shared by the existing SQL command handlers."""
+
+from azure.cli.core.aaz import (
+    AAZBoolArg, AAZBoolType, AAZCommand, AAZHttpOperation, AAZIntArg,
+    AAZIntType, AAZObjectType, AAZStrArg, AAZStrType,
+)
+
+
+class _ShortTermRetentionPolicy(AAZCommand):
+    def _handler(self, command_args):
+        # None means omitted in the existing Python handlers, not an explicit JSON null.
+        super()._handler({key: value for key, value in command_args.items() if value is not None})
+
+    @classmethod
+    def _build_arguments_schema(cls, *args, **kwargs):
+        schema = super()._build_arguments_schema(*args, **kwargs)
+        schema.resource_group_name = AAZStrArg(required=True)
+        schema.parent_name = AAZStrArg(required=True)
+        schema.database_name = AAZStrArg(required=True)
+        schema.managed = AAZBoolArg(default=False)
+        schema.deleted = AAZBoolArg(default=False)
+        schema.retention_days = AAZIntArg()
+        schema.diffbackup_hours = AAZIntArg()
+        schema.lock_immutability = AAZBoolArg()
+        return schema
+
+    def _output(self, _result=None):
+        return self.deserialize_output(self.ctx.vars.instance, client_flatten=True)
+
+
+class ShortTermRetentionPolicyGet(_ShortTermRetentionPolicy):
+    def _handler(self, command_args):
+        super()._handler(command_args)
+        _PolicyGet(ctx=self.ctx)()
+        return self._output()
+
+
+class ShortTermRetentionPolicySet(_ShortTermRetentionPolicy):
+    AZ_SUPPORT_NO_WAIT = True
+
+    def _handler(self, command_args):
+        super()._handler(command_args)
+        return self.build_lro_poller(self._execute_operations, self._output)
+
+    def _execute_operations(self):
+        yield _PolicySet(ctx=self.ctx)()
+
+
+class _PolicyOperation(AAZHttpOperation):  # pylint: disable=abstract-method
+    CLIENT_TYPE = "MgmtClient"
+
+    @property
+    def url(self):
+        parent_type = "managedInstances" if self.ctx.args.managed else "servers"
+        child_type = "restorableDroppedDatabases" if self.ctx.args.deleted else "databases"
+        return self.client.format_url(
+            "/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}"
+            "/providers/Microsoft.Sql/" + parent_type + "/{parentName}/" + child_type +
+            "/{databaseName}/backupShortTermRetentionPolicies/default",
+            **self.url_parameters
+        )
+
+    @property
+    def url_parameters(self):
+        return {
+            **self.serialize_url_param("subscriptionId", self.ctx.subscription_id, required=True),
+            **self.serialize_url_param("resourceGroupName", self.ctx.args.resource_group_name, required=True),
+            **self.serialize_url_param("parentName", self.ctx.args.parent_name, required=True),
+            **self.serialize_url_param("databaseName", self.ctx.args.database_name, required=True),
+        }
+
+    @property
+    def query_parameters(self):
+        return {"api-version": "2026-08-01-preview"}
+
+    @property
+    def header_parameters(self):
+        return {"Accept": "application/json", "Content-Type": "application/json"}
+
+    @property
+    def error_format(self):
+        return "ODataV4Format"
+
+    def on_success(self, session):
+        self.ctx.set_var(
+            "instance", self.deserialize_http_content(session),
+            schema_builder=self._build_response_schema,
+        )
+
+    @staticmethod
+    def _build_response_schema():
+        schema = AAZObjectType()
+        schema.id = AAZStrType(flags={"read_only": True})
+        schema.name = AAZStrType(flags={"read_only": True})
+        schema.type = AAZStrType(flags={"read_only": True})
+        schema.properties = AAZObjectType(flags={"client_flatten": True})
+        properties = schema.properties
+        properties.retention_days = AAZIntType(serialized_name="retentionDays")
+        properties.diff_backup_interval_in_hours = AAZIntType(serialized_name="diffBackupIntervalInHours")
+        properties.lock_immutability = AAZBoolType(serialized_name="lockImmutability")
+        properties.immutability_status = AAZStrType(
+            serialized_name="immutabilityStatus", flags={"read_only": True},
+        )
+        schema.system_data = AAZObjectType(serialized_name="systemData", flags={"read_only": True})
+        system_data = schema.system_data
+        system_data.created_by = AAZStrType(serialized_name="createdBy")
+        system_data.created_by_type = AAZStrType(serialized_name="createdByType")
+        system_data.created_at = AAZStrType(serialized_name="createdAt")
+        system_data.last_modified_by = AAZStrType(serialized_name="lastModifiedBy")
+        system_data.last_modified_by_type = AAZStrType(serialized_name="lastModifiedByType")
+        system_data.last_modified_at = AAZStrType(serialized_name="lastModifiedAt")
+        return schema
+
+
+class _PolicyGet(_PolicyOperation):
+    @property
+    def method(self):
+        return "GET"
+
+    def __call__(self, *args, **kwargs):
+        session = self.client.send_request(request=self.make_request(), stream=False, **kwargs)
+        if session.http_response.status_code == 200:
+            return self.on_success(session)
+        return self.on_error(session.http_response)
+
+
+class _PolicySet(_PolicyOperation):
+    @property
+    def method(self):
+        return "PUT"
+
+    @property
+    def content(self):
+        value, builder = self.new_content_builder(self.ctx.args, typ=AAZObjectType)
+        builder.set_prop("properties", AAZObjectType)
+        properties = builder.get(".properties")
+        properties.set_prop("retentionDays", AAZIntType, ".retention_days")
+        if not self.ctx.args.managed:
+            properties.set_prop("diffBackupIntervalInHours", AAZIntType, ".diffbackup_hours")
+        if not self.ctx.args.deleted:
+            properties.set_prop("lockImmutability", AAZBoolType, ".lock_immutability")
+        return self.serialize_content(value)
+
+    def __call__(self, *args, **kwargs):
+        session = self.client.send_request(request=self.make_request(), stream=False, **kwargs)
+        if session.http_response.status_code in (200, 202):
+            return self.client.build_lro_polling(
+                self.ctx.args.no_wait, session, self.on_success, self.on_error,
+                lro_options={"final-state-via": "location"},
+                path_format_arguments=self.url_parameters,
+            )
+        return self.on_error(session.http_response)

src/azure-cli/azure/cli/command_modules/sql/commands.py

@@ -26,7 +26,6 @@
 )
 
 from ._util import (
-    get_sql_backup_short_term_retention_policies_operations,
     get_sql_server_azure_ad_administrators_operations,
     get_sql_capabilities_operations,
     get_sql_databases_operations,
@@ -55,7 +54,6 @@
     get_sql_recoverable_managed_databases_operations,
     get_sql_managed_database_advanced_threat_protection_settings_operations,
     get_sql_managed_database_restore_details_operations,
-    get_sql_managed_backup_short_term_retention_policies_operations,
     get_sql_managed_database_long_term_retention_policies_operations,
     get_sql_managed_database_long_term_retention_backups_operations,
     get_sql_managed_instance_advanced_threat_protection_settings_operations,
@@ -384,17 +382,11 @@ def load_command_table(self, _):
             'restore',
             'restore_geo_backup')
 
-    backup_short_term_retention_policies_operations = CliCommandType(
-        operations_tmpl='azure.mgmt.sql.operations#BackupShortTermRetentionPoliciesOperations.{}',
-        client_factory=get_sql_backup_short_term_retention_policies_operations)
-
-    with self.command_group('sql db str-policy',
-                            backup_short_term_retention_policies_operations,
-                            client_factory=get_sql_backup_short_term_retention_policies_operations) as g:
+    with self.command_group('sql db str-policy') as g:
 
         g.custom_command('set', 'update_short_term_retention', supports_no_wait=True)
         g.custom_show_command('show', 'get_short_term_retention')
-        g.wait_command('wait')
+        g.custom_wait_command('wait', 'get_short_term_retention')
 
     database_sensitivity_labels_operations = CliCommandType(
         operations_tmpl='azure.mgmt.sql.operations#SensitivityLabelsOperations.{}',
@@ -951,13 +943,7 @@ def load_command_table(self, _):
                                  setter_type=managed_database_advanced_threat_protection_setting_update_sdk,
                                  custom_func_name='midb_advanced_threat_protection_setting_update')
 
-    managed_backup_short_term_retention_policies_operations = CliCommandType(
-        operations_tmpl='azure.mgmt.sql.operations#ManagedBackupShortTermRetentionPoliciesOperations.{}',
-        client_factory=get_sql_managed_backup_short_term_retention_policies_operations)
-
-    with self.command_group('sql midb short-term-retention-policy',
-                            managed_backup_short_term_retention_policies_operations,
-                            client_factory=get_sql_managed_backup_short_term_retention_policies_operations) as g:
+    with self.command_group('sql midb short-term-retention-policy') as g:
 
         g.custom_command(
             'set',