enables backup retention and collation setting for SQL DB

Author: dzsquared

.changes/unreleased/added-sqldatabase-creationpayload.yaml

@@ -0,0 +1,6 @@
+kind: added
+body: Add SQL Database creationPayload support for mkdir command with mode, backupRetentionDays, and collation parameters
+time: 2026-03-13T00:00:00.000000000Z
+custom:
+    Author: dzsquared
+    AuthorLink: https://github.com/dzsquared

src/fabric_cli/utils/fab_cmd_mkdir_utils.py

@@ -188,6 +188,27 @@ def add_type_specific_payload(item: Item, args, payload):
                 ]
             }
 
+        case ItemType.SQL_DATABASE:
+            _mode = params.get("mode")
+            _backup_retention_days = params.get("backupretentiondays")
+            _collation = params.get("collation")
+
+            if _mode or _backup_retention_days or _collation:
+                creation_payload: dict = {"creationMode": _mode if _mode else "new"}
+                if _backup_retention_days:
+                    try:
+                        creation_payload["backupRetentionDays"] = int(
+                            _backup_retention_days
+                        )
+                    except ValueError:
+                        raise FabricCLIError(
+                            f"'backupRetentionDays' must be a valid integer, got '{_backup_retention_days}'",
+                            fab_constant.ERROR_INVALID_INPUT,
+                        )
+                if _collation:
+                    creation_payload["collation"] = _collation
+                payload_dict["creationPayload"] = creation_payload
+
     return payload_dict
 
 
@@ -309,6 +330,8 @@ def get_params_per_item_type(item: Item):
             optional_params = ["semanticModelId"]
         case ItemType.MOUNTED_DATA_FACTORY:
             required_params = ["subscriptionId", "resourceGroup", "factoryName"]
+        case ItemType.SQL_DATABASE:
+            optional_params = ["mode", "backupRetentionDays", "collation"]
 
     return required_params, optional_params
 

tests/test_commands/conftest.py

@@ -83,6 +83,8 @@
         ["Latin1_General_100_CI_AS_KS_WS_SC_UTF8"]),
     (ItemType.WAREHOUSE, "", ["Latin1_General_100_BIN2_UTF8"]),
     (ItemType.REPORT, "", ["_auto"]),
+    (ItemType.SQL_DATABASE, "backupRetentionDays=21,collation=SQL_Latin1_General_CP1_CS_AS",
+        ["properties"]),
 ])
 
 mv_item_to_item_success_params = pytest.mark.parametrize("item_type", [

tests/test_commands/test_mkdir.py

@@ -286,6 +286,184 @@ def test_mkdir_item_with_creation_payload_success(
         # Cleanup
         rm(item_full_path)
 
+    def test_mkdir_sqldatabase_without_params_success(
+        self,
+        workspace,
+        cli_executor,
+        mock_print_done,
+        mock_questionary_print,
+        vcr_instance,
+        cassette_name,
+        upsert_item_to_cache,
+    ):
+        # Setup
+        sqldb_display_name = generate_random_string(vcr_instance, cassette_name)
+        sqldb_full_path = cli_path_join(
+            workspace.full_path, f"{sqldb_display_name}.{ItemType.SQL_DATABASE}"
+        )
+
+        # Execute command without -P flag
+        cli_executor.exec_command(f"mkdir {sqldb_full_path}")
+
+        # Assert creation success
+        upsert_item_to_cache.assert_called_once()
+        mock_print_done.assert_called_once()
+        assert sqldb_display_name in mock_print_done.call_args[0][0]
+
+        # Verify item was created without creationPayload
+        mock_questionary_print.reset_mock()
+        get(sqldb_full_path, query=".")
+        mock_questionary_print.assert_called_once()
+        result_output = mock_questionary_print.call_args[0][0]
+        assert sqldb_display_name in result_output
+        assert "creationPayload" not in result_output
+
+        # Cleanup
+        rm(sqldb_full_path)
+
+    def test_mkdir_sqldatabase_partial_params_success(
+        self,
+        workspace,
+        cli_executor,
+        mock_print_done,
+        mock_questionary_print,
+        vcr_instance,
+        cassette_name,
+        upsert_item_to_cache,
+    ):
+        # Setup
+        sqldb_display_name = generate_random_string(vcr_instance, cassette_name)
+        sqldb_full_path = cli_path_join(
+            workspace.full_path, f"{sqldb_display_name}.{ItemType.SQL_DATABASE}"
+        )
+
+        # Execute command with only backupRetentionDays (no collation)
+        cli_executor.exec_command(f"mkdir {sqldb_full_path} -P backupRetentionDays=21")
+
+        # Assert creation success
+        upsert_item_to_cache.assert_called_once()
+        mock_print_done.assert_called_once()
+        assert sqldb_display_name in mock_print_done.call_args[0][0]
+
+        # Verify item was created with properties
+        mock_questionary_print.reset_mock()
+        get(sqldb_full_path, query=".")
+        mock_questionary_print.assert_called_once()
+        result_output = mock_questionary_print.call_args[0][0]
+        assert sqldb_display_name in result_output
+
+        # Cleanup
+        rm(sqldb_full_path)
+
+    def test_mkdir_sqldatabase_invalid_retention_days_failure(
+        self,
+        workspace,
+        cli_executor,
+        assert_fabric_cli_error,
+        vcr_instance,
+        cassette_name,
+    ):
+        # Setup
+        sqldb_display_name = generate_random_string(vcr_instance, cassette_name)
+        sqldb_full_path = cli_path_join(
+            workspace.full_path, f"{sqldb_display_name}.{ItemType.SQL_DATABASE}"
+        )
+
+        # Execute command with invalid backupRetentionDays
+        cli_executor.exec_command(f"mkdir {sqldb_full_path} -P backupRetentionDays=abc")
+
+        # Assert failure due to invalid integer
+        assert_fabric_cli_error(constant.ERROR_INVALID_INPUT)
+
+    def test_mkdir_sqldatabase_explicit_mode_success(
+        self,
+        workspace,
+        cli_executor,
+        mock_print_done,
+        mock_questionary_print,
+        vcr_instance,
+        cassette_name,
+        upsert_item_to_cache,
+    ):
+        # Setup
+        sqldb_display_name = generate_random_string(vcr_instance, cassette_name)
+        sqldb_full_path = cli_path_join(
+            workspace.full_path, f"{sqldb_display_name}.{ItemType.SQL_DATABASE}"
+        )
+
+        # Execute command with explicit mode
+        cli_executor.exec_command(f"mkdir {sqldb_full_path} -P mode=new")
+
+        # Assert creation success
+        upsert_item_to_cache.assert_called_once()
+        mock_print_done.assert_called_once()
+        assert sqldb_display_name in mock_print_done.call_args[0][0]
+
+        # Verify item was created with properties
+        mock_questionary_print.reset_mock()
+        get(sqldb_full_path, query=".")
+        mock_questionary_print.assert_called_once()
+        result_output = mock_questionary_print.call_args[0][0]
+        assert sqldb_display_name in result_output
+
+        # Cleanup
+        rm(sqldb_full_path)
+
+    def test_mkdir_sqldatabase_mode_only_success(
+        self,
+        workspace,
+        cli_executor,
+        mock_print_done,
+        mock_questionary_print,
+        vcr_instance,
+        cassette_name,
+        upsert_item_to_cache,
+    ):
+        # Setup
+        sqldb_display_name = generate_random_string(vcr_instance, cassette_name)
+        sqldb_full_path = cli_path_join(
+            workspace.full_path, f"{sqldb_display_name}.{ItemType.SQL_DATABASE}"
+        )
+
+        # Execute command with only mode param
+        cli_executor.exec_command(f"mkdir {sqldb_full_path} -P mode=new")
+
+        # Assert creation success
+        upsert_item_to_cache.assert_called_once()
+        mock_print_done.assert_called_once()
+        assert sqldb_display_name in mock_print_done.call_args[0][0]
+
+        # Verify item was created
+        mock_questionary_print.reset_mock()
+        get(sqldb_full_path, query=".")
+        mock_questionary_print.assert_called_once()
+        result_output = mock_questionary_print.call_args[0][0]
+        assert sqldb_display_name in result_output
+
+        # Cleanup
+        rm(sqldb_full_path)
+
+    def test_mkdir_sqldatabase_param_discovery(
+        self, workspace, cli_executor, mock_questionary_print
+    ):
+        # Setup
+        sqldb_full_path = cli_path_join(
+            workspace.full_path, f"paramDiscovery.{ItemType.SQL_DATABASE}"
+        )
+
+        # Execute command with -P but no value
+        cli_executor.exec_command(f"mkdir {sqldb_full_path} -P")
+
+        # Assert param discovery message is printed
+        no_params_message = (
+            "Params for '.SQLDatabase'. Use key=value separated by commas."
+        )
+        output = mock_questionary_print.call_args[0][0]
+        assert no_params_message in output
+        assert "backupRetentionDays" in output
+        assert "collation" in output
+        assert "mode" in output
+
     def test_mkdir_mounted_data_factory_with_required_params_success(
         self,
         workspace,

tests/test_utils/test_fab_cmd_mkdir_utils.py

@@ -10,9 +10,12 @@
 from fabric_cli.core.fab_exceptions import FabricCLIError
 from fabric_cli.errors import ErrorMessages
 from fabric_cli.utils.fab_cmd_mkdir_utils import (
+    add_type_specific_payload,
     find_mpe_connection,
     get_connection_config_from_params,
+    get_params_per_item_type,
 )
+from fabric_cli.core.fab_types import ItemType
 
 
 def test_fabric_data_pipelines_workspace_identity_no_params_success():
@@ -209,4 +212,147 @@ def test_find_mpe_connection_return_403_success(self):
             called_url = call_args.args[1] if len(call_args.args) > 1 else call_args.kwargs['url']
             assert "privateEndpointConnections" in called_url
             assert "api-version=2023-11-01" in called_url
+
+
+class TestGetParamsPerItemTypeSqlDatabase:
+    """Test cases for get_params_per_item_type with SQL_DATABASE."""
+
+    def test_sql_database_returns_optional_params(self):
+        """Test that SQL_DATABASE returns the correct optional params."""
+        mock_item = Mock()
+        mock_item.item_type = ItemType.SQL_DATABASE
+
+        required, optional = get_params_per_item_type(mock_item)
+
+        assert required == []
+        assert optional == ["mode", "backupRetentionDays", "collation"]
+
+    def test_sql_database_has_no_required_params(self):
+        """Test that SQL_DATABASE has no required params."""
+        mock_item = Mock()
+        mock_item.item_type = ItemType.SQL_DATABASE
+
+        required, _ = get_params_per_item_type(mock_item)
+
+        assert len(required) == 0
+
+
+class TestAddTypeSpecificPayloadSqlDatabase:
+    """Test cases for add_type_specific_payload with SQL_DATABASE."""
+
+    def _make_item_and_args(self, params):
+        """Helper to create mock item and args for SQL_DATABASE."""
+        mock_item = Mock()
+        mock_item.item_type = ItemType.SQL_DATABASE
+        mock_args = Namespace(params=params)
+        return mock_item, mock_args
+
+    def test_all_params_success(self):
+        """Test SQL_DATABASE with all params provided."""
+        item, args = self._make_item_and_args(
+            {
+                "mode": "new",
+                "backupretentiondays": "21",
+                "collation": "SQL_Latin1_General_CP1_CI_AS",
+            }
+        )
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert "creationPayload" in result
+        cp = result["creationPayload"]
+        assert cp["creationMode"] == "new"
+        assert cp["backupRetentionDays"] == 21
+        assert cp["collation"] == "SQL_Latin1_General_CP1_CI_AS"
+
+    def test_partial_params_backup_only_success(self):
+        """Test SQL_DATABASE with only backupRetentionDays provided."""
+        item, args = self._make_item_and_args({"backupretentiondays": "7"})
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert "creationPayload" in result
+        cp = result["creationPayload"]
+        assert cp["creationMode"] == "new"  # Default
+        assert cp["backupRetentionDays"] == 7
+        assert "collation" not in cp
+
+    def test_partial_params_collation_only_success(self):
+        """Test SQL_DATABASE with only collation provided."""
+        item, args = self._make_item_and_args(
+            {"collation": "SQL_Latin1_General_CP1_CI_AS"}
+        )
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert "creationPayload" in result
+        cp = result["creationPayload"]
+        assert cp["creationMode"] == "new"  # Default
+        assert "backupRetentionDays" not in cp
+        assert cp["collation"] == "SQL_Latin1_General_CP1_CI_AS"
+
+    def test_no_recognized_params_no_creation_payload(self):
+        """Test SQL_DATABASE with no recognized params produces no creationPayload."""
+        item, args = self._make_item_and_args({"unrelated": "value"})
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert "creationPayload" not in result
+
+    def test_empty_params_no_creation_payload(self):
+        """Test SQL_DATABASE with empty params dict produces no creationPayload."""
+        item, args = self._make_item_and_args({})
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert "creationPayload" not in result
+
+    def test_invalid_backup_retention_days_failure(self):
+        """Test SQL_DATABASE with non-integer backupRetentionDays raises FabricCLIError."""
+        item, args = self._make_item_and_args({"backupretentiondays": "abc"})
+        payload = {"displayName": "testdb"}
+
+        with pytest.raises(FabricCLIError) as exc_info:
+            add_type_specific_payload(item, args, payload)
+
+        assert "backupRetentionDays" in str(exc_info.value.message)
+        assert "abc" in str(exc_info.value.message)
+        assert exc_info.value.status_code == fab_constant.ERROR_INVALID_INPUT
+
+    def test_explicit_mode_success(self):
+        """Test SQL_DATABASE with explicit mode value."""
+        item, args = self._make_item_and_args({"mode": "copy"})
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert result["creationPayload"]["creationMode"] == "copy"
+
+    def test_mode_only_success(self):
+        """Test SQL_DATABASE with only mode provided."""
+        item, args = self._make_item_and_args({"mode": "new"})
+        payload = {"displayName": "testdb"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        cp = result["creationPayload"]
+        assert cp["creationMode"] == "new"
+        assert "backupRetentionDays" not in cp
+        assert "collation" not in cp
+
+    def test_preserves_existing_payload_fields(self):
+        """Test that SQL_DATABASE creation doesn't remove existing payload fields."""
+        item, args = self._make_item_and_args({"mode": "new"})
+        payload = {"displayName": "testdb", "description": "A test database"}
+
+        result = add_type_specific_payload(item, args, payload)
+
+        assert result["displayName"] == "testdb"
+        assert result["description"] == "A test database"
+        assert "creationPayload" in result