Merge pull request #1233 from gooddata/snapshot-master-81da27bd-to-rel/dev

[bot] Merge master/81da27bd into rel/dev

Author: yenkins-admin

packages/gooddata-pipelines/src/gooddata_pipelines/api/gooddata_sdk.py

@@ -52,20 +52,6 @@ def check_workspace_exists(self, workspace_id: str) -> bool:
         except Exception:
             return False
 
-    def get_workspace(self, workspace_id: str, **_: str) -> CatalogWorkspace:
-        """
-        Calls GoodData Python SDK to retrieve a workspace by its ID.
-
-        Args:
-            workspace_id (str): The ID of the workspace to retrieve.
-        Returns:
-            CatalogWorkspace: The workspace object retrieved from the SDK.
-        Raises:
-            GoodDataApiException: If the workspace cannot be retrieved, an exception
-            is raised with additional context information.
-        """
-        return self._sdk.catalog_workspace.get_workspace(workspace_id)
-
     def delete_panther_workspace(self, workspace_id: str) -> None:
         """
         Calls GoodData Python SDK to delete a workspace by its ID.
@@ -184,20 +170,6 @@ def put_declarative_permissions(
             workspace_id, ws_permissions
         )
 
-    def get_user(self, user_id: str, **_: str) -> CatalogUser:
-        """
-        Calls GoodData Python SDK to retrieve a user by its ID.
-
-        Args:
-            user_id (str): The ID of the user to retrieve.
-        Returns:
-            CatalogUser: The user object retrieved from the SDK.
-        Raises:
-            GoodDataApiException: If the user cannot be retrieved, an exception
-            is raised with additional context information.
-        """
-        return self._sdk.catalog_user.get_user(user_id)
-
     def create_or_update_user(self, user: CatalogUser, **_: str) -> None:
         """
         Calls GoodData Python SDK to create or update a user.
@@ -226,20 +198,6 @@ def delete_user(self, user_id: str, **_: str) -> None:
         """
         return self._sdk.catalog_user.delete_user(user_id)
 
-    def get_user_group(self, user_group_id: str, **_: str) -> CatalogUserGroup:
-        """
-        Calls GoodData Python SDK to retrieve a user group by its ID.
-
-        Args:
-            user_group_id (str): The ID of the user group to retrieve.
-        Returns:
-            CatalogUserGroup: The user group object retrieved from the SDK.
-        Raises:
-            GoodDataApiException: If the user group cannot be retrieved, an exception
-            is raised with additional context information.
-        """
-        return self._sdk.catalog_user.get_user_group(user_group_id)
-
     def list_user_groups(self) -> list[CatalogUserGroup]:
         """
         Calls GoodData Python SDK to retrieve all user groups.

packages/gooddata-pipelines/src/gooddata_pipelines/provisioning/entities/users/permissions.py

@@ -4,7 +4,8 @@
 
 from typing import TypeVar
 
-from gooddata_pipelines.api.exceptions import GoodDataApiException
+from gooddata_api_client.exceptions import NotFoundException  # type: ignore
+
 from gooddata_pipelines.provisioning.entities.users.models.permissions import (
     EntityType,
     PermissionDeclaration,
@@ -14,7 +15,6 @@
     WSPermissionsDeclarations,
 )
 from gooddata_pipelines.provisioning.provisioning import Provisioning
-from gooddata_pipelines.provisioning.utils.exceptions import BaseUserException
 
 # Type variable for permission models (PermissionIncrementalLoad or PermissionFullLoad)
 PermissionModel = TypeVar(
@@ -109,20 +109,32 @@ def _check_user_group_exists(self, ug_id: str) -> None:
 
     def _validate_permission(
         self, permission: PermissionFullLoad | PermissionIncrementalLoad
-    ) -> None:
-        """Validates if the permission is correctly defined."""
-        if permission.entity_type == EntityType.user:
-            self._api.get_user(
-                permission.entity_id, error_message="User not found"
+    ) -> bool:
+        """Validates that all entities referenced in the permission exist.
+
+        Raises:
+            RuntimeError: If an unexpected error is encountered.
+        """
+        all_entities_exist = True
+        try:
+            if permission.entity_type == EntityType.user:
+                self._api._sdk.catalog_user.get_user(permission.entity_id)
+            else:
+                self._api._sdk.catalog_user.get_user_group(permission.entity_id)
+
+            self._api._sdk.catalog_workspace.get_workspace(
+                permission.workspace_id
             )
-        else:
-            self._api.get_user_group(
-                permission.entity_id, error_message="User group not found"
+
+        except NotFoundException:
+            all_entities_exist = False
+
+        except Exception as e:
+            raise RuntimeError(
+                f"Failed to validate permission: {e.__class__.__name__}: {e}"
             )
 
-        self._api.get_workspace(
-            permission.workspace_id, error_message="Workspace not found"
-        )
+        return all_entities_exist
 
     def _filter_invalid_permissions(
         self,
@@ -132,11 +144,9 @@ def _filter_invalid_permissions(
         valid_permissions: list[PermissionModel] = []
 
         for permission in permissions:
-            try:
-                self._validate_permission(permission)
-            except (BaseUserException, GoodDataApiException) as e:
+            if not self._validate_permission(permission):
                 self.logger.error(
-                    f"Skipping {permission}. Error: {e.error_message} "
+                    f"Skipping {permission}. Error: Permission references non-existent entities."
                     + f"Context: {permission.__dict__}"
                 )
                 continue