refactor(set): refactor the update flow to construct PATCH request bodies correctly

.changes/unreleased/optimization-20260101-115402.yaml

@@ -0,0 +1,3 @@
+kind: optimization
+body: set command - refactor the update flow to construct PATCH request bodies by extracting only the updated properties from the GET payload.
+time: 2026-01-01T11:54:02.344718831Z

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

@@ -29,22 +29,12 @@ def exec(virtual_ws_item: VirtualWorkspaceItem, args: Namespace) -> None:
         args.output = None
         vwsi_capacity_def = get_capacity.exec(virtual_ws_item, args, verbose=False)
 
-        json_payload, updated_def = utils_set.update_fabric_element(
-            vwsi_capacity_def, query, args.input, decode_encode=False
+        updated_def = utils_set.update_fabric_element(
+            vwsi_capacity_def, query, args.input
         )
 
-        def _prep_for_updated_def(data):
-            data.pop("id", None)
-            data.pop("type", None)
-            data.pop("name", None)
-            data.pop("tags", None)
-            data.pop("fabricId", None)
-            return json.dumps(data, indent=4)
-
-        capacity_update_def = _prep_for_updated_def(updated_def)
-
         utils_ui.print_grey(f"Setting new property for '{virtual_ws_item.name}'...")
-        response = capacity_api.update_capacity(args, capacity_update_def)
+        response = capacity_api.update_capacity(args, json.dumps(updated_def))
 
         if response.status_code == 200:
             utils_ui.print_output_format(args, message="Capacity updated")

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

@@ -12,6 +12,7 @@
 from fabric_cli.utils import fab_ui as utils_ui
 
 JMESPATH_UPDATE_CONNECTIONS = ["displayName", "privacyLevel", "credentialDetails"]
+CONECTIVITY_TYPE_KEY = "connectivityType"
 
 
 def exec(connection: VirtualWorkspaceItem, args: Namespace) -> None:
@@ -25,18 +26,15 @@ def exec(connection: VirtualWorkspaceItem, args: Namespace) -> None:
         args.deep_traversal = True
         args.output = None
         vwsi_connection_def = get_connection.exec(connection, args, verbose=False)
+        connectivity_type = vwsi_connection_def.get(CONECTIVITY_TYPE_KEY, "")
 
-        json_payload, updated_def = utils_set.update_fabric_element(
-            vwsi_connection_def, query, args.input, decode_encode=False
+        updated_def = utils_set.update_fabric_element(
+            vwsi_connection_def, query, args.input
         )
 
         def _prep_for_updated_def(data):
-            data.pop("id", None)  # Remove 'id' if it exists
-            data.pop("gatewayId", None)  # Remove 'type' if it exists
-            data.pop(
-                "connectionDetails", None
-            )  # Remove 'connectionDetails' if it exists
-            return json.dumps(data, indent=4)
+            data[CONECTIVITY_TYPE_KEY] = connectivity_type
+            return json.dumps(data)
 
         connection_update_def = _prep_for_updated_def(updated_def)
 
@@ -45,8 +43,7 @@ def _prep_for_updated_def(data):
         response = connection_api.update_connection(args, connection_update_def)
 
         if response.status_code == 200:
-            # Update mem_store
-            connection._name = updated_def["displayName"]
-            utils_mem_store.upsert_connection_to_cache(connection)
-
+            utils_set.update_cache(
+                updated_def, connection, utils_mem_store.upsert_connection_to_cache
+            )
             utils_ui.print_output_format(args, message="Connection updated")

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

@@ -26,28 +26,18 @@ def exec(virtual_ws_item: VirtualWorkspaceItem, args: Namespace) -> None:
         args.output = None
         vwsi_domain_def = get_domain.exec(virtual_ws_item, args, verbose=False)
 
-        _, updated_def = utils_set.update_fabric_element(
-            vwsi_domain_def, query, args.input, decode_encode=False
+        updated_def = utils_set.update_fabric_element(
+            vwsi_domain_def, query, args.input
         )
 
-        def _prep_for_updated_def(data):
-            data.pop("id", None)  # Remove 'id' if it exists
-            data.pop("type", None)  # Remove 'type' if it exists
-            data.pop("name", None)  # Remove 'name' if it exists
-            data.pop("tags", None)  # Remove 'tags' if it exists
-            return json.dumps(data, indent=4)
-
-        domain_update_def = _prep_for_updated_def(updated_def)
         args.name = virtual_ws_item.short_name
         args.id = virtual_ws_item.id
 
         utils_ui.print_grey(f"Setting new property for '{virtual_ws_item.name}'...")
-        response = domain_api.update_domain(args, domain_update_def)
+        response = domain_api.update_domain(args, json.dumps(updated_def))
 
         if response.status_code == 200:
-            # Update mem_store
-            new_domain_name = updated_def["displayName"]
-            virtual_ws_item._name = new_domain_name
-            utils_mem_store.upsert_domain_to_cache(virtual_ws_item)
-
+            utils_set.update_cache(
+                updated_def, virtual_ws_item, utils_mem_store.upsert_domain_to_cache
+            )
             utils_ui.print_output_format(args, message="Domain updated")

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

@@ -26,26 +26,16 @@ def exec(folder: Folder, args: Namespace) -> None:
         args.output = None
         folder_def = get_folder.exec(folder, args, verbose=False)
 
-        _, updated_def = utils_set.update_fabric_element(
-            folder_def, query, args.input, decode_encode=False
-        )
+        updated_def = utils_set.update_fabric_element(folder_def, query, args.input)
 
-        def _prep_for_updated_def(data):
-            data.pop("id", None)  # Remove 'id' if it exists
-            data.pop("workspaceId", None)  # Remove 'workspaceId' if it exists
-            data.pop("parentFolderId", None)  # Remove 'parentFolderId' if it exists
-            return json.dumps(data, indent=4)
-
-        folder_update_def = _prep_for_updated_def(updated_def)
         args.name = folder.short_name
         args.id = folder.id
 
         utils_ui.print_grey(f"Setting new property for '{folder.name}'...")
-        response = folder_api.update_folder(args, folder_update_def)
+        response = folder_api.update_folder(args, json.dumps(updated_def))
 
         if response.status_code == 200:
-            # Update mem_store
-            folder._name = updated_def["displayName"]
-            utils_mem_store.upsert_folder_to_cache(folder)
-
+            utils_set.update_cache(
+                updated_def, folder, utils_mem_store.upsert_folder_to_cache
+            )
             utils_ui.print_output_format(args, message="Folder updated")

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

@@ -9,6 +9,7 @@
 from fabric_cli.core import fab_constant
 from fabric_cli.core.fab_exceptions import FabricCLIError
 from fabric_cli.core.hiearchy.fab_hiearchy import VirtualWorkspaceItem
+from fabric_cli.errors import ErrorMessages
 from fabric_cli.utils import fab_cmd_set_utils as utils_set
 from fabric_cli.utils import fab_mem_store as utils_mem_store
 from fabric_cli.utils import fab_ui as utils_ui
@@ -22,6 +23,8 @@
     "numberOfMemberGateways",
 ]
 
+SUPPORTED_GATEWAY_TYPES = ["OnPremises", "VirtualNetwork"]
+
 
 def exec(gateway: VirtualWorkspaceItem, args: Namespace) -> None:
     query = args.query
@@ -35,25 +38,30 @@ def exec(gateway: VirtualWorkspaceItem, args: Namespace) -> None:
         args.output = None
         vwsi_gateway_def = get_gateway.exec(gateway, args, verbose=False)
 
-        json_payload, updated_def = utils_set.update_fabric_element(
-            vwsi_gateway_def, query, args.input, decode_encode=False
+        gatewat_type = vwsi_gateway_def.get("type", "")
+
+        if gatewat_type not in SUPPORTED_GATEWAY_TYPES:
+            raise FabricCLIError(
+                ErrorMessages.Common.gateway_type_not_supported(gatewat_type),
+                fab_constant.ERROR_NOT_SUPPORTED,
+            )
+        elif gatewat_type == "OnPremises" and query.startswith(
+            ("numberOfMemberGateways", "capacityId", "inactivityMinutesBeforeSleep")
+        ):
+            raise FabricCLIError(
+                ErrorMessages.Common.gateway_property_not_supported_for_type(
+                    query, "OnPremises"
+                ),
+                fab_constant.ERROR_NOT_SUPPORTED,
+            )
+
+        updated_def = utils_set.update_fabric_element(
+            vwsi_gateway_def, query, args.input
         )
 
-        def _prep_for_updated_def(data):
-            data.pop("id", None)
-            # numberOfMemberGateways is supported only for VirtualNetwork type (reason for the whole match statement)
-            match data.get("type"):
-                case "OnPremises":
-                    data.pop("numberOfMemberGateways", None)
-                    data.pop("publicKey", None)
-                    data.pop("version", None)
-                case "VirtualNetwork":
-                    data.pop("virtualNetworkAzureResource", None)
-                case _:
-                    raise FabricCLIError(
-                        f"Set Operation on Gateway type '{data.get('type')}' not supported",
-                        fab_constant.ERROR_NOT_SUPPORTED,
-                    )
+        def _prep_for_updated_def(data, gatewat_type: str) -> str:
+            data["type"] = gatewat_type
+
             # Casting to int if the value is a string and present
             if isinstance(data.get("inactivityMinutesBeforeSleep", 0), str):
                 data["inactivityMinutesBeforeSleep"] = int(
@@ -62,16 +70,16 @@ def _prep_for_updated_def(data):
             if isinstance(data.get("numberOfMemberGateways", 0), str):
                 data["numberOfMemberGateways"] = int(data["numberOfMemberGateways"])
 
-            return json.dumps(data, indent=4)
+            return json.dumps(data)
 
-        gateway_update_def = _prep_for_updated_def(updated_def)
+        gateway_update_def = _prep_for_updated_def(updated_def, gatewat_type)
 
         args.id = gateway.id
         utils_ui.print_grey(f"Setting new property for '{gateway.name}'...")
         response = gateways_api.update_gateway(args, gateway_update_def)
 
         if response.status_code == 200:
-            # Update mem_store
-            gateway._name = updated_def["displayName"]
-            utils_mem_store.upsert_gateway_to_cache(gateway)
+            utils_set.update_cache(
+                updated_def, gateway, utils_mem_store.upsert_gateway_to_cache
+            )
             utils_ui.print_output_format(args, message="Gateway updated")

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

@@ -38,24 +38,17 @@ def exec(item: Item, args: Namespace) -> None:
             def_response = item_api.get_item_definition(args)
             definition = json.loads(def_response.text)
 
-            updated_def = _update_element(
-                definition, query_value, args.input, decode_encode=True
-            )
+            updated_def = _update_item_definition(definition, query_value, args.input)
 
-            definition_base64_to_update, _ = utils_set.extract_json_schema(updated_def)
-            update_item_definition_payload = json.dumps(definition_base64_to_update)
+            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)
         else:
             item_metadata = json.loads(item_api.get_item(args, item_uri=True).text)
 
-            updated_metadata = _update_element(
-                item_metadata, query_value, args.input, decode_encode=False
-            )
-
-            update_payload_dict = utils_set.extract_updated_properties(
-                updated_metadata, query_value
+            update_payload_dict = _update_item_metadata(
+                item_metadata, query_value, args.input
             )
             item_update_payload = json.dumps(update_payload_dict)
 
@@ -64,28 +57,46 @@ def exec(item: Item, args: Namespace) -> None:
             item_api.update_item(args, item_update_payload, item_uri=True)
 
             if fab_constant.ITEM_QUERY_DISPLAY_NAME in update_payload_dict:
-                new_item_name = updated_metadata[fab_constant.ITEM_QUERY_DISPLAY_NAME]
+                new_item_name = update_payload_dict[
+                    fab_constant.ITEM_QUERY_DISPLAY_NAME
+                ]
                 item._name = new_item_name
                 utils_mem_store.upsert_item_to_cache(item)
 
         utils_ui.print_output_format(args, message="Item updated")
 
 
-def _update_element(
-    resource_def: dict,
+def _update_item_definition(
+    item_def: dict,
     query_value: str,
     input_value: str,
-    decode_encode: bool,
 ) -> dict:
     try:
-        _, updated_def = utils_set.update_fabric_element(
-            resource_def,
+        updated_def = utils_set.update_item_definition(
+            item_def,
             query_value,
             input_value,
-            decode_encode=decode_encode,
         )
         return updated_def
-    except (ValueError, KeyError, IndexError):
+    except Exception:
+        raise FabricCLIError(
+            CommonErrors.invalid_set_item_query(query_value),
+            fab_constant.ERROR_INVALID_QUERY,
+        )
+
+
+def _update_item_metadata(
+    item_metadata: dict,
+    query_value: str,
+    input_value: str,
+) -> dict:
+    try:
+        return utils_set.update_fabric_element(
+            item_metadata,
+            query_value,
+            input_value,
+        )
+    except Exception:
         raise FabricCLIError(
             CommonErrors.invalid_set_item_query(query_value),
             fab_constant.ERROR_INVALID_QUERY,

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

@@ -29,8 +29,9 @@ def onelake_shortcut(onelake: OneLakeItem, args: Namespace) -> None:
     utils_set.print_set_warning()
     if args.force or utils_ui.prompt_confirm():
 
-        _, updated_def = utils_set.update_fabric_element(
-            shortcut_def, query, args.input, decode_encode=False
+        # Read new values from the user and retrieve updated shortcut definition with the new values.
+        updated_def = utils_set.update_fabric_element(
+            shortcut_def, query, args.input, return_full_element=True
         )
 
         if "target" in updated_def and "type" in updated_def["target"]:

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

@@ -31,14 +31,14 @@ def exec(virtual_item: VirtualItem, args: Namespace) -> None:
         args.output = None
         vi_spark_pool_def = get_sparkpool.exec(virtual_item, args, verbose=False)
 
-        json_payload, updated_def = utils_set.update_fabric_element(
-            vi_spark_pool_def, query, args.input, decode_encode=False
+        updated_def = utils_set.update_fabric_element(
+            vi_spark_pool_def, query, args.input
         )
 
         def _prep_for_updated_def(data):
             data.pop("id", None)  # Remove 'id' if it exists
             data.pop("type", None)  # Remove 'type' if it exists
-            return json.dumps(data, indent=4)
+            return json.dumps(data)
 
         spark_pool_update_def = _prep_for_updated_def(updated_def)
 
@@ -48,8 +48,11 @@ def _prep_for_updated_def(data):
         response = sparkpool_api.update_spark_pool(args, spark_pool_update_def)
 
         if response.status_code == 200:
-            # Update mem_store
-            virtual_item._name = updated_def["name"]
-            utils_mem_store.upsert_spark_pool_to_cache(virtual_item)
+            utils_set.update_cache(
+                updated_def,
+                virtual_item,
+                utils_mem_store.upsert_spark_pool_to_cache,
+                element_name_key="name",
+            )
 
             utils_ui.print_output_format(args, message="Spark Pool updated")

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

@@ -31,16 +31,9 @@ def exec(workspace: Workspace, args: Namespace) -> None:
     utils_set.print_set_warning()
     if args.force or utils_ui.prompt_confirm():
 
-        json_payload, updated_def = utils_set.update_fabric_element(
-            workspace_def, query, args.input, decode_encode=False
-        )
-
-        definition_base64_to_update, name_description_properties = (
-            utils_set.extract_json_schema(updated_def, definition=False)
-        )
+        updated_def = utils_set.update_fabric_element(workspace_def, query, args.input)
 
         args.ws_id = workspace.id
-        update_workspace_payload = json.dumps(name_description_properties)
 
         utils_ui.print_grey(f"Setting new property for '{workspace.name}'...")
 
@@ -53,11 +46,10 @@ def exec(workspace: Workspace, args: Namespace) -> None:
             )
         # Update workspace
         else:
-            response = workspace_api.update_workspace(args, update_workspace_payload)
+            response = workspace_api.update_workspace(args, json.dumps(updated_def))
 
         if response.status_code == 200:
-            # Update mem_store
-            workspace._name = name_description_properties["displayName"]
-            utils_mem_store.upsert_workspace_to_cache(workspace)
-
+            utils_set.update_cache(
+                updated_def, workspace, utils_mem_store.upsert_workspace_to_cache
+            )
             utils_ui.print_output_format(args, message="Workspace updated")