refactor(core): consolidate format-building logic in utils (#195)

Co-authored-by: Alon Yeshurun <alonyeshurun+microsoft@microsoft.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

Author: 4 people

src/fabric_cli/client/fab_api_item.py

@@ -100,7 +100,9 @@ def get_item(
 
 def get_item_definition(args: Namespace) -> ApiResponse:
     """https://learn.microsoft.com/en-us/rest/api/fabric/core/items/get-item-definition"""
-    args.uri = f"workspaces/{args.ws_id}/items/{args.id}/getDefinition{args.format}"
+    args.uri = f"workspaces/{args.ws_id}/items/{args.id}/getDefinition"
+    if args.format:
+        args.uri += f"?format={args.format}"
     args.method = "post"
     args.wait = True
 

src/fabric_cli/commands/fs/export/fab_fs_export_item.py

@@ -10,10 +10,9 @@
 from fabric_cli.core import fab_constant
 from fabric_cli.core.fab_commands import Command
 from fabric_cli.core.fab_exceptions import FabricCLIError
-from fabric_cli.core.fab_types import ItemType, definition_format_mapping
+from fabric_cli.core.fab_types import ItemType
 from fabric_cli.core.hiearchy.fab_folder import Folder
 from fabric_cli.core.hiearchy.fab_hiearchy import Item, Workspace
-from fabric_cli.errors import ErrorMessages
 from fabric_cli.utils import fab_cmd_export_utils as utils_export
 from fabric_cli.utils import fab_item_util, fab_mem_store, fab_storage, fab_ui
 
@@ -103,27 +102,8 @@ def export_single_item(
         args.from_path = item.path.strip("/")
         args.ws_id, args.id, args.item_type = workspace_id, item_id, str(item_type)
 
-        # Get definition_format_mapping for item without default fallback
-        valid_export_formats = definition_format_mapping.get(item_type, {})
-        # Get export_format_param from args without default
         export_format_param = getattr(args, "format", None)
-        
-        if export_format_param not in valid_export_formats:
-             # Export format not in definition_format_mapping
-            if not export_format_param:
-                # Empty format param - use default formats if exists
-                args.format = valid_export_formats.get("default", "")
-            else:
-                # Non-empty format param but not supported
-                available_formats = [k for k in valid_export_formats.keys() if k != "default"]
-                raise FabricCLIError(
-                    ErrorMessages.Export.invalid_export_format(available_formats),
-                    fab_constant.ERROR_INVALID_INPUT,
-                )
-        else:
-            # Export format is explicitly supported
-            args.format = valid_export_formats[export_format_param]
-           
+        args.format = fab_item_util.resolve_definition_format(item_type, export_format_param)
 
         item_def = item_api.get_item_withdefinition(args, item_uri)
 

src/fabric_cli/commands/fs/impor/fab_fs_import_item.py

@@ -8,33 +8,19 @@
 from fabric_cli.client.fab_api_types import ApiResponse
 from fabric_cli.core import fab_constant, fab_logger
 from fabric_cli.core.fab_exceptions import FabricCLIError
-from fabric_cli.core.fab_types import ItemType, definition_format_mapping
+from fabric_cli.core.fab_types import ItemType
 from fabric_cli.core.hiearchy.fab_hiearchy import Item
 from fabric_cli.utils import fab_cmd_import_utils as utils_import
+from fabric_cli.utils import fab_item_util
 from fabric_cli.utils import fab_mem_store as utils_mem_store
 from fabric_cli.utils import fab_storage as utils_storage
 from fabric_cli.utils import fab_ui as utils_ui
 
 
 def import_single_item(item: Item, args: Namespace) -> None:
-    _input_format = None
-    if args.format:
-        _input_format = args.format
-        if item.item_type in definition_format_mapping:
-            valid_formats = list(
-                definition_format_mapping[item.item_type].keys())
-            if _input_format not in valid_formats:
-                available_formats = [
-                    k for k in valid_formats if k != "default"]
-                raise FabricCLIError(
-                    f"Invalid format. Only the following formats are supported: {', '.join(available_formats)}",
-                    fab_constant.ERROR_INVALID_INPUT,
-                )
-        else:
-            raise FabricCLIError(
-                f"Invalid format. No formats are supported",
-                fab_constant.ERROR_INVALID_INPUT,
-            )
+    _input_format = fab_item_util.resolve_definition_format(
+        item_type=item.item_type, format_param=getattr(args, "format", None)
+    )
 
     args.ws_id = item.workspace.id
     input_path = utils_storage.get_import_path(args.input)
@@ -54,7 +40,7 @@ def import_single_item(item: Item, args: Namespace) -> None:
 
         # Get the payload
         payload = utils_import.get_payload_for_item_type(
-            _input_path, item, _input_format
+            _input_path, item, input_format=_input_format
         )
 
         if item_exists:

src/fabric_cli/commands/fs/set/fab_fs_set_item.py

@@ -33,19 +33,24 @@ def exec(item: Item, args: Namespace) -> None:
         args.item_uri = format_mapping.get(item.item_type, "items")
 
         if query_value.startswith(fab_constant.ITEM_QUERY_DEFINITION):
-            formats = definition_format_mapping.get(item.item_type, {"default": ""})
+            formats = definition_format_mapping.get(
+                item.item_type, {"default": ""})
+            # plain value; query param built in get_item_definition()
             args.format = formats["default"]
             def_response = item_api.get_item_definition(args)
             definition = json.loads(def_response.text)
 
-            updated_def = _update_item_definition(definition, query_value, args.input)
+            updated_def = _update_item_definition(
+                definition, query_value, args.input)
 
             update_item_definition_payload = json.dumps(updated_def)
 
             utils_ui.print_grey(f"Setting new property for '{item.name}'...")
-            item_api.update_item_definition(args, update_item_definition_payload)
+            item_api.update_item_definition(
+                args, update_item_definition_payload)
         else:
-            item_metadata = json.loads(item_api.get_item(args, item_uri=True).text)
+            item_metadata = json.loads(
+                item_api.get_item(args, item_uri=True).text)
 
             update_payload_dict = _update_item_metadata(
                 item_metadata, query_value, args.input

src/fabric_cli/core/fab_types.py

@@ -583,19 +583,19 @@ class MirroredDatabaseFolders(Enum):
 
 definition_format_mapping = {
     ItemType.SPARK_JOB_DEFINITION: {
-        "default": "?format=SparkJobDefinitionV1",
-        "SparkJobDefinitionV1": "?format=SparkJobDefinitionV1",
-        "SparkJobDefinitionV2": "?format=SparkJobDefinitionV2",
+        "default": "SparkJobDefinitionV1",
+        "SparkJobDefinitionV1": "SparkJobDefinitionV1",
+        "SparkJobDefinitionV2": "SparkJobDefinitionV2",
     },
     ItemType.NOTEBOOK: {
-        "default": "?format=ipynb",
-        ".py": "?format=fabricGitSource",
-        ".ipynb": "?format=ipynb",
+        "default": "ipynb",
+        ".py": "fabricGitSource",
+        ".ipynb": "ipynb",
     },
     ItemType.SEMANTIC_MODEL: {
         "default": "",
-        "TMDL": "?format=TMDL",
-        "TMSL": "?format=TMSL",
+        "TMDL": "TMDL",
+        "TMSL": "TMSL",
     },
     ItemType.COSMOS_DB_DATABASE: {"default": ""},
     ItemType.USER_DATA_FUNCTION: {"default": ""},

src/fabric_cli/core/hiearchy/fab_item.py

@@ -74,89 +74,5 @@ def workspace(self) -> Workspace:
             assert isinstance(self.parent, Folder)
             return self.parent.workspace
 
-    def get_payload(self, definition, input_format=None) -> dict:
-        match self.item_type:
-
-            case ItemType.SPARK_JOB_DEFINITION:
-                return {
-                    "type": str(self.item_type),
-                    "description": "Imported from fab",
-                    "folderId": self.folder_id,
-                    "displayName": self.short_name,
-                    "definition": {
-                        "format": (
-                            "SparkJobDefinitionV1"
-                            if input_format is None
-                            else input_format
-                        ),
-                        "parts": definition["parts"],
-                    },
-                }
-            case ItemType.NOTEBOOK:
-                return {
-                    "type": str(self.item_type),
-                    "description": "Imported from fab",
-                    "folderId": self.folder_id,
-                    "displayName": self.short_name,
-                    "definition": {
-                        **(
-                            {"parts": definition["parts"]}
-                            if input_format == ".py"
-                            else {"format": "ipynb", "parts": definition["parts"]}
-                        )
-                    },
-                }
-            case ItemType.SEMANTIC_MODEL:
-                return {
-                    "type": str(self.item_type),
-                    "description": "Imported from fab",
-                    "folderId": self.folder_id,
-                    "displayName": self.short_name,
-                    "definition": (
-                        definition
-                        if input_format is None
-                        else {
-                            "format": input_format,
-                            "parts": definition["parts"],
-                        }
-                    ),
-                }
-            case (
-                ItemType.REPORT
-                | ItemType.KQL_DASHBOARD
-                | ItemType.DATA_PIPELINE
-                | ItemType.KQL_QUERYSET
-                | ItemType.EVENTHOUSE
-                | ItemType.KQL_DATABASE
-                | ItemType.MIRRORED_DATABASE
-                | ItemType.DIGITAL_TWIN_BUILDER
-                | ItemType.REFLEX
-                | ItemType.EVENTSTREAM
-                | ItemType.MOUNTED_DATA_FACTORY
-                | ItemType.COPYJOB
-                | ItemType.VARIABLE_LIBRARY
-                | ItemType.GRAPHQLAPI
-                | ItemType.DATAFLOW
-                | ItemType.SQL_DATABASE
-                | ItemType.COSMOS_DB_DATABASE
-                | ItemType.GRAPH_QUERY_SET
-                | ItemType.USER_DATA_FUNCTION
-                | ItemType.MAP
-            ):
-                return {
-                    "type": str(self.item_type),
-                    "description": "Imported from fab",
-                    "folderId": self.folder_id,
-                    "displayName": self.short_name,
-                    "definition": definition,
-                }
-            case _:
-                raise FabricCLIError(
-                    ErrorMessages.Hierarchy.item_type_doesnt_support_definition_payload(
-                        str(self.item_type)
-                    ),
-                    fab_constant.ERROR_UNSUPPORTED_COMMAND,
-                )
-
     def get_folders(self) -> List[str]:
         return ItemFoldersMap.get(self.item_type, [])

src/fabric_cli/errors/__init__.py

@@ -7,7 +7,6 @@
 from .config import ConfigErrors
 from .context import ContextErrors
 from .cp import CpErrors
-from .export import ExportErrors
 from .hierarchy import HierarchyErrors
 from .labels import LabelsErrors
 from .mkdir import MkdirErrors
@@ -23,7 +22,6 @@ class ErrorMessages:
     Config = ConfigErrors
     Context = ContextErrors
     Cp = CpErrors
-    Export = ExportErrors
     Hierarchy = HierarchyErrors
     Labels = LabelsErrors
     Mkdir = MkdirErrors

src/fabric_cli/errors/common.py

@@ -248,3 +248,12 @@ def gateway_property_not_supported_for_type(
     @staticmethod
     def query_not_supported_for_set(query: str) -> str:
         return f"Query '{query}' is not supported for set command"
+
+    @staticmethod
+    def invalid_definition_format(valid_formats: list[str]) -> str:
+        if valid_formats:
+            message = f"Only the following formats are supported: {', '.join(valid_formats)}"
+        else:
+            message = "No formats are supported"
+        return f"Invalid format. {message}"
+

src/fabric_cli/errors/export.py

@@ -25,11 +25,17 @@ def get_payload_for_item_type(
     if item.item_type == ItemType.ENVIRONMENT:
         return _build_environment_payload(path)
     else:
-        base64_definition = _build_payload(path)
-        return item.get_payload(base64_definition, input_format)
+        definition = _build_definition(path, input_format)
+        return {
+            "type": str(item.item_type),
+            "description": "Imported from fab",
+            "folderId": item.folder_id,
+            "displayName": item.short_name,
+            "definition": definition,
+        }
 
 
-def _build_payload(input_path: Any) -> dict:
+def _build_definition(input_path: Any, input_format: Optional[str] = None) -> dict:
     directory = input_path
     parts = []
 
@@ -67,9 +73,10 @@ def _build_payload(input_path: Any) -> dict:
                 }
             )
 
-    # Create the final JSON structure
-    payload_structure = {"parts": parts}
-    return payload_structure
+    definition: dict = {"parts": parts}
+    if input_format:
+        definition["format"] = input_format
+    return definition
 
 
 def _encode_file_to_base64(file_path: str) -> str: