added shortcut smart diff feature

Author: Christoffer Bäckström

.changes/unreleased/added-20260401-135149.yaml

@@ -0,0 +1,8 @@
+kind: added
+body: Add feature flag for shortcut smart diff publishing
+time: 2026-04-01T13:51:49.2071983+02:00
+custom:
+    Author: bckstrm
+    AuthorLink: https://github.com/bckstrm
+    Issue: "904"
+    IssueLink: https://github.com/microsoft/fabric-cicd/issues/904

src/fabric_cicd/_items/_lakehouse.py

@@ -3,6 +3,7 @@
 
 """Functions to process and deploy Lakehouse item."""
 
+import copy
 import json
 import logging
 
@@ -57,28 +58,39 @@ def check_sqlendpoint_provision_status(fabric_workspace_obj: FabricWorkspace, it
         iteration += 1
 
 
-def list_deployed_shortcuts(fabric_workspace_obj: FabricWorkspace, item_obj: Item) -> list:
+def _get_shortcut_path(shortcut: dict) -> str:
     """
-    Lists all deployed shortcut paths
+    Build a unique shortcut path from shortcut metadata.
+
+    Args:
+        shortcut: The shortcut definition dictionary
+    """
+    return f"{shortcut['path']}/{shortcut['name']}"
+
+
+def list_deployed_shortcuts(fabric_workspace_obj: FabricWorkspace, item_obj: Item) -> dict[str, dict]:
+    """
+    Lists all deployed shortcut by path
 
     Args:
         fabric_workspace_obj: The FabricWorkspace object containing the items to be published
         item_obj: The item object to list the shortcuts for
     """
     request_url = f"{fabric_workspace_obj.base_api_url}/items/{item_obj.guid}/shortcuts"
-    deployed_shortcut_paths = []
+    deployed_shortcuts_by_path = {}
 
     while request_url:
         # https://learn.microsoft.com/en-us/rest/api/fabric/core/onelake-shortcuts/list-shortcuts
         response = fabric_workspace_obj.endpoint.invoke(method="GET", url=request_url)
 
         # Handle cases where the response body is empty
         shortcuts = response["body"].get("value", [])
-        deployed_shortcut_paths.extend(f"{shortcut['path']}/{shortcut['name']}" for shortcut in shortcuts)
+        for shortcut in shortcuts:
+            deployed_shortcuts_by_path[_get_shortcut_path(shortcut)] = shortcut
 
         request_url = response["header"].get("continuationUri", None)
 
-    return deployed_shortcut_paths
+    return deployed_shortcuts_by_path
 
 
 def replace_default_lakehouse_id(shortcut: dict, item_obj: Item) -> dict:
@@ -165,6 +177,43 @@ def _unpublish_shortcuts(self, shortcut_paths: list) -> None:
                 url=f"{self.fabric_workspace_obj.base_api_url}/items/{self.item_obj.guid}/shortcuts/{deployed_shortcut_path}",
             )
 
+    @staticmethod
+    def _is_shortcut_subset_match(repository_shortcut: object, deployed_shortcut: object) -> bool:
+        """
+        Recursively compare shortcut fields while ignoring extra deployed fields.
+
+        This allows server-side schema extensions without forcing unnecessary republish.
+        """
+        if isinstance(repository_shortcut, dict):
+            if not isinstance(deployed_shortcut, dict):
+                return False
+            for key, repository_value in repository_shortcut.items():
+                if key not in deployed_shortcut:
+                    return False
+                if not ShortcutPublisher._is_shortcut_subset_match(repository_value, deployed_shortcut[key]):
+                    return False
+            return True
+
+        if isinstance(repository_shortcut, list):
+            if not isinstance(deployed_shortcut, list) or len(repository_shortcut) != len(deployed_shortcut):
+                return False
+            return all(
+                ShortcutPublisher._is_shortcut_subset_match(repository_item, deployed_item)
+                for repository_item, deployed_item in zip(repository_shortcut, deployed_shortcut)
+            )
+
+        return repository_shortcut == deployed_shortcut
+
+    def _is_shortcut_changed(self, shortcut: dict, deployed_shortcut: dict) -> bool:
+        """
+        Determine whether a shortcut needs publishing.
+
+        Returns:
+            True when the shortcut differs from deployed state and should be published.
+        """
+        shortcut_for_compare = replace_default_lakehouse_id(copy.deepcopy(shortcut), self.item_obj)
+        return not self._is_shortcut_subset_match(shortcut_for_compare, deployed_shortcut)
+
     def publish_one(self, _shortcut_name: str, shortcut: dict) -> None:
         """
         Publish a single shortcut.
@@ -230,12 +279,29 @@ def publish_all(self) -> None:
                 )
                 logger.info(f"{constants.INDENT}Excluded shortcuts: {excluded_shortcuts}")
 
-        shortcuts_to_publish = {f"{shortcut['path']}/{shortcut['name']}": shortcut for shortcut in shortcuts}
+        shortcuts_to_publish = {_get_shortcut_path(shortcut): shortcut for shortcut in shortcuts}
 
         if shortcuts_to_publish:
             logger.info(f"Publishing Lakehouse '{self.item_obj.name}' Shortcuts")
-            shortcut_paths_to_unpublish = [path for path in deployed_shortcuts if path not in shortcuts_to_publish]
+            shortcut_paths_to_unpublish = [
+                path for path in list(deployed_shortcuts.keys()) if path not in shortcuts_to_publish
+            ]
             self._unpublish_shortcuts(shortcut_paths_to_unpublish)
-            # Deploy and overwrite shortcuts
+            if FeatureFlag.ENABLE_SHORTCUT_SMART_DIFF.value in constants.FEATURE_FLAG:
+                shortcuts_with_change = {}
+                unchanged_shortcut_count = 0
+                for shortcut_path, shortcut in shortcuts_to_publish.items():
+                    deployed_shortcut = deployed_shortcuts.get(shortcut_path)
+                    if deployed_shortcut and not self._is_shortcut_changed(shortcut, deployed_shortcut):
+                        unchanged_shortcut_count += 1
+                        continue
+                    shortcuts_with_change[shortcut_path] = shortcut
+                shortcuts_to_publish = shortcuts_with_change
+                if unchanged_shortcut_count:
+                    logger.info(
+                        f"{constants.INDENT}Skipped {unchanged_shortcut_count} unchanged shortcut(s) via smart diff"
+                    )
+
+            # Deploy and overwrite changed/new shortcuts
             for shortcut_path, shortcut in shortcuts_to_publish.items():
                 self.publish_one(shortcut_path, shortcut)

src/fabric_cicd/constants.py

@@ -117,6 +117,8 @@ class FeatureFlag(str, Enum):
     """Set to enable the deletion of KQL Databases (attached to Eventhouses)."""
     ENABLE_SHORTCUT_PUBLISH = "enable_shortcut_publish"
     """Set to enable deploying shortcuts with the lakehouse."""
+    ENABLE_SHORTCUT_SMART_DIFF = "enable_shortcut_smart_diff"
+    """Set to enable change comparision before deploying shortcuts"""
     DISABLE_WORKSPACE_FOLDER_PUBLISH = "disable_workspace_folder_publish"
     """Set to disable deploying workspace sub folders."""
     CONTINUE_ON_SHORTCUT_FAILURE = "continue_on_shortcut_failure"

tests/test_deploy_with_config.py

@@ -267,6 +267,7 @@ def test_extract_parameter_file_path_missing(self):
         settings = extract_workspace_settings(config, "dev")
         assert "parameter_file_path" not in settings
 
+
 class TestPublishSettingsExtraction:
     """Test publish settings extraction from config."""
 
@@ -1001,6 +1002,7 @@ def test_deploy_with_config_loads_parameter_when_field_present(self, tmp_path):
             call_kwargs = mock_fabric_ws.call_args[1]
             assert call_kwargs.get("skip_parameterization") is False
 
+
 class TestConfigIntegration:
     """Integration tests for config functionality."""
 

tests/test_shortcut_exclude.py

@@ -8,6 +8,7 @@
 
 import pytest
 
+from fabric_cicd import constants
 from fabric_cicd._common._item import Item
 from fabric_cicd._items._lakehouse import ShortcutPublisher
 from fabric_cicd.fabric_workspace import FabricWorkspace
@@ -56,6 +57,39 @@ def create_shortcut_file(shortcuts_data):
     return file_obj
 
 
+def set_deployed_shortcuts(mock_fabric_workspace, deployed_shortcuts):
+    """Configure endpoint behavior to return specific deployed shortcuts."""
+
+    def mock_invoke(method, url, **_kwargs):
+        if method == "GET" and "shortcuts" in url:
+            return {"body": {"value": deployed_shortcuts}, "header": {}}
+        if method == "POST" and "shortcuts" in url:
+            return {"body": {"id": "mock-shortcut-id"}}
+        if method == "DELETE" and "shortcuts" in url:
+            return {"body": {}}
+        return {"body": {}}
+
+    mock_fabric_workspace.endpoint.invoke.side_effect = mock_invoke
+
+
+def get_shortcut_post_calls(mock_fabric_workspace):
+    """Return all shortcut create/overwrite API calls."""
+    return [
+        call
+        for call in mock_fabric_workspace.endpoint.invoke.call_args_list
+        if call[1].get("method") == "POST" and "shortcuts" in call[1].get("url", "")
+    ]
+
+
+@pytest.fixture
+def restore_feature_flags():
+    """Ensure feature flags are restored after test."""
+    original_flags = constants.FEATURE_FLAG.copy()
+    yield
+    constants.FEATURE_FLAG.clear()
+    constants.FEATURE_FLAG.update(original_flags)
+
+
 def test_process_shortcuts_with_exclude_regex_filters_shortcuts(mock_fabric_workspace, mock_item):
     """Test that shortcut_exclude_regex correctly filters shortcuts from deployment."""
 
@@ -304,3 +338,140 @@ def test_process_shortcuts_with_complex_regex_pattern(mock_fabric_workspace, moc
     # Verify the published shortcut is the prod one
     published_shortcut = post_calls[0][1]["body"]
     assert published_shortcut["name"] == "prod_shortcut"
+
+
+def test_shortcut_smart_diff_skips_unchanged_shortcuts(mock_fabric_workspace, mock_item):
+    """Smart diff should skip publish for unchanged shortcuts."""
+    shortcuts_data = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+        },
+        {
+            "name": "shortcut2",
+            "path": "/Files",
+            "target": {"type": "OneLake", "oneLake": {"path": "Files/s2", "itemId": "item-2"}},
+        },
+    ]
+    deployed_shortcuts = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+            "serverManaged": {"createdBy": "system"},
+        },
+        {
+            "name": "shortcut2",
+            "path": "/Files",
+            "target": {"type": "OneLake", "oneLake": {"path": "Files/s2", "itemId": "item-2"}},
+            "serverManaged": {"createdBy": "system"},
+        },
+    ]
+
+    mock_item.item_files = [create_shortcut_file(shortcuts_data)]
+    set_deployed_shortcuts(mock_fabric_workspace, deployed_shortcuts)
+    constants.FEATURE_FLAG.add("enable_shortcut_smart_diff")
+    try:
+        ShortcutPublisher(mock_fabric_workspace, mock_item).publish_all()
+        post_calls = get_shortcut_post_calls(mock_fabric_workspace)
+        assert len(post_calls) == 0
+    finally:
+        constants.FEATURE_FLAG.discard("enable_shortcut_smart_diff")
+
+
+def test_shortcut_smart_diff_publishes_only_changed_shortcut(mock_fabric_workspace, mock_item):
+    """Smart diff should publish only shortcuts with changed fields."""
+    shortcuts_data = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+        },
+        {
+            "name": "shortcut2",
+            "path": "/Files",
+            "target": {"type": "OneLake", "oneLake": {"path": "Files/s2-new", "itemId": "item-2"}},
+        },
+    ]
+    deployed_shortcuts = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+        },
+        {
+            "name": "shortcut2",
+            "path": "/Files",
+            "target": {"type": "OneLake", "oneLake": {"path": "Files/s2-old", "itemId": "item-2"}},
+        },
+    ]
+
+    mock_item.item_files = [create_shortcut_file(shortcuts_data)]
+    set_deployed_shortcuts(mock_fabric_workspace, deployed_shortcuts)
+    constants.FEATURE_FLAG.add("enable_shortcut_smart_diff")
+    try:
+        ShortcutPublisher(mock_fabric_workspace, mock_item).publish_all()
+
+        post_calls = get_shortcut_post_calls(mock_fabric_workspace)
+        assert len(post_calls) == 1
+        assert post_calls[0][1]["body"]["name"] == "shortcut2"
+    finally:
+        constants.FEATURE_FLAG.discard("enable_shortcut_smart_diff")
+
+
+def test_shortcut_smart_diff_publishes_client_side_new_field(mock_fabric_workspace, mock_item):
+    """Smart diff should publish when repo contains a new field missing from deployed shortcut."""
+    shortcuts_data = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+            "clientField": {"owner": "team-a"},
+        }
+    ]
+    deployed_shortcuts = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+        }
+    ]
+
+    mock_item.item_files = [create_shortcut_file(shortcuts_data)]
+    set_deployed_shortcuts(mock_fabric_workspace, deployed_shortcuts)
+    constants.FEATURE_FLAG.add("enable_shortcut_smart_diff")
+    try:
+        ShortcutPublisher(mock_fabric_workspace, mock_item).publish_all()
+
+        post_calls = get_shortcut_post_calls(mock_fabric_workspace)
+        assert len(post_calls) == 1
+        assert post_calls[0][1]["body"]["name"] == "shortcut1"
+    finally:
+        constants.FEATURE_FLAG.discard("enable_shortcut_smart_diff")
+
+
+def test_shortcut_smart_diff_disabled_keeps_publish_all_behavior(mock_fabric_workspace, mock_item):
+    """When smart diff is disabled, unchanged shortcuts are still published."""
+    shortcuts_data = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+        }
+    ]
+    deployed_shortcuts = [
+        {
+            "name": "shortcut1",
+            "path": "/Tables",
+            "target": {"type": "OneLake", "oneLake": {"path": "Tables/s1", "itemId": "item-1"}},
+        }
+    ]
+
+    mock_item.item_files = [create_shortcut_file(shortcuts_data)]
+    set_deployed_shortcuts(mock_fabric_workspace, deployed_shortcuts)
+
+    ShortcutPublisher(mock_fabric_workspace, mock_item).publish_all()
+
+    post_calls = get_shortcut_post_calls(mock_fabric_workspace)
+    assert len(post_calls) == 1