diff --git a/packages/gooddata-pipelines/src/gooddata_pipelines/api/exceptions.py b/packages/gooddata-pipelines/src/gooddata_pipelines/api/exceptions.py deleted file mode 100644 index 4328b6f21..000000000 --- a/packages/gooddata-pipelines/src/gooddata_pipelines/api/exceptions.py +++ /dev/null @@ -1,41 +0,0 @@ -# (C) 2025 GoodData Corporation - -"""Exception class for Panther operations. - -This module defines the internally used `PantherException` class, which is used -to handle exceptions that occur during operations related to the Panther SDK or -GoodData Cloud API. -""" - - -class GoodDataApiException(Exception): - """Exception raised during Panther operations. - - This exception is used to indicate errors that occur during operations - related to interactions with the GoodData Python SDK or GoodData Cloud API. - It can include additional context provided through keyword arguments. - """ - - def __init__(self, message: str, **kwargs: str) -> None: - """Raise a PantherException with a message and optional context. - - Args: - message (str): The error message describing the exception. - **kwargs: Additional context for the exception, such as HTTP status, - API endpoint, and HTTP method or any other relevant information. - """ - - super().__init__(message) - self.error_message: str = message - - # Set default values for attributes. - # TODO: Consider if the defaults for these are still needed - # - the values were necessary for log schema implementations, which - # are not used anymore. - self.http_status: str = "500 Internal Server Error" - self.api_endpoint: str = "NA" - self.http_method: str = "NA" - - # Set attributes from kwargs. - for key, value in kwargs.items(): - setattr(self, key, value) diff --git a/packages/gooddata-pipelines/src/gooddata_pipelines/api/gooddata_sdk.py b/packages/gooddata-pipelines/src/gooddata_pipelines/api/gooddata_sdk.py deleted file mode 100644 index d9e6af5f9..000000000 --- a/packages/gooddata-pipelines/src/gooddata_pipelines/api/gooddata_sdk.py +++ /dev/null @@ -1,332 +0,0 @@ -# (C) 2025 GoodData Corporation - -"""Interaction with GoodData Cloud via the Gooddata Python SDK.""" - -from pathlib import Path -from typing import Callable - -from gooddata_sdk.catalog.permission.declarative_model.permission import ( - CatalogDeclarativeWorkspacePermissions, -) -from gooddata_sdk.catalog.user.entity_model.user import CatalogUser -from gooddata_sdk.catalog.user.entity_model.user_group import CatalogUserGroup -from gooddata_sdk.catalog.workspace.declarative_model.workspace.workspace import ( - CatalogDeclarativeWorkspaceDataFilters, -) -from gooddata_sdk.catalog.workspace.entity_model.user_data_filter import ( - CatalogUserDataFilter, -) -from gooddata_sdk.catalog.workspace.entity_model.workspace import ( - CatalogWorkspace, -) -from gooddata_sdk.sdk import GoodDataSdk - -from gooddata_pipelines.api.utils import raise_with_context - - -def apply_to_all_methods(decorator: Callable) -> Callable: - def decorate(cls: type) -> type: - for attr in cls.__dict__: - if callable(getattr(cls, attr)) and not attr.startswith("__"): - setattr(cls, attr, decorator(getattr(cls, attr))) - return cls - - return decorate - - -@apply_to_all_methods(raise_with_context()) -class SdkMethods: - """ - Class to intaract with GoodData Cloud via the Gooddata Python SDK. - """ - - _sdk: GoodDataSdk - - def get_organization_id(self) -> str: - return self._sdk.catalog_organization.organization_id - - def check_workspace_exists(self, workspace_id: str) -> bool: - try: - self._sdk.catalog_workspace.get_workspace(workspace_id) - return True - except Exception: - return False - - def delete_panther_workspace(self, workspace_id: str) -> None: - """ - Calls GoodData Python SDK to delete a workspace by its ID. - - Args: - workspace_id (str): The ID of the workspace to delete. - Raises: - GoodDataApiException: If the workspace cannot be deleted, an exception - is raised with additional context information. - """ - self._sdk.catalog_workspace.delete_workspace(workspace_id) - - def create_or_update_panther_workspace( - self, - workspace_id: str, - workspace_name: str, - parent_id: str | None, - **_: str, - ) -> None: - """ - Calls GoodData Python SDK to create or update a workspace with the given ID, - name, and parent ID. - - Args: - workspace_id (str): The ID of the workspace to create or update. - workspace_name (str): The name of the workspace. - parent_id (str | None): The ID of the parent workspace, if any. - Returns: - None - Raises: - GoodDataApiException: If the workspace cannot be created or updated, - an exception is raised with additional context information. - """ - return self._sdk.catalog_workspace.create_or_update( - CatalogWorkspace( - workspace_id=workspace_id, - name=workspace_name, - parent_id=parent_id, - ) - ) - - def get_panther_children_workspaces( - self, parent_workspace_ids: set[str] - ) -> list[CatalogWorkspace]: - """ - Calls GoodData Python SDK to retrieve all workspaces in domain and filters the - result by the set of parent workspace IDs. - - Args: - parent_workspace_ids (set[str]): A set of parent workspace IDs to filter - child workspaces. - Returns: - list[CatalogWorkspace]: List of child workspaces in the parent workspace. - """ - all_workspaces: list[CatalogWorkspace] = self.list_workspaces() - - children: list[CatalogWorkspace] = [ - workspace - for workspace in all_workspaces - if workspace.parent_id in parent_workspace_ids - ] - - return children - - def list_workspaces(self) -> list[CatalogWorkspace]: - """Retrieves all workspaces in the GoodData Cloud domain. - - Returns: - list[CatalogWorkspace]: A list of all workspaces in the domain. - Raises: - GoodDataApiException: If the workspaces cannot be retrieved, an exception - is raised with additional context information. - """ - return self._sdk.catalog_workspace.list_workspaces() - - def get_declarative_permissions( - self, workspace_id: str - ) -> CatalogDeclarativeWorkspacePermissions: - """ - Retrieves the declarative permissions for a given workspace. - - Args: - workspace_id (str): The ID of the workspace for which to retrieve - permissions. - Returns: - CatalogDeclarativeWorkspacePermissions: The declarative permissions - for the workspace. - Raises: - GoodDataApiException: If the permissions cannot be retrieved, an exception - is raised with additional context information. - """ - return self._sdk.catalog_permission.get_declarative_permissions( - workspace_id - ) - - def put_declarative_permissions( - self, - workspace_id: str, - ws_permissions: CatalogDeclarativeWorkspacePermissions, - ) -> None: - """ - Updates the declarative permissions for a given workspace. - - Args: - workspace_id (str): The ID of the workspace for which to update - permissions. - ws_permissions (CatalogDeclarativeWorkspacePermissions): The new - declarative permissions to set for the workspace. - Returns: - None - Raises: - GoodDataApiException: If the permissions cannot be updated, an exception - is raised with additional context information. - """ - return self._sdk.catalog_permission.put_declarative_permissions( - workspace_id, ws_permissions - ) - - def create_or_update_user(self, user: CatalogUser, **_: str) -> None: - """ - Calls GoodData Python SDK to create or update a user. - - Args: - user (CatalogUser): The user object to create or update. - Returns: - None - Raises: - GoodDataApiException: If the user cannot be created or updated, - an exception is raised with additional context information. - """ - return self._sdk.catalog_user.create_or_update_user(user) - - def delete_user(self, user_id: str, **_: str) -> None: - """ - Calls GoodData Python SDK to delete a user by its ID. - - Args: - user_id (str): The ID of the user to delete. - Returns: - None - Raises: - GoodDataApiException: If the user cannot be deleted, an exception - is raised with additional context information. - """ - return self._sdk.catalog_user.delete_user(user_id) - - def list_user_groups(self) -> list[CatalogUserGroup]: - """ - Calls GoodData Python SDK to retrieve all user groups. - - Returns: - list[CatalogUserGroup]: A list of all user groups in the domain. - Raises: - GoodDataApiException: If the user groups cannot be retrieved, an - exception is raised with additional context information. - """ - return self._sdk.catalog_user.list_user_groups() - - def list_users(self) -> list[CatalogUser]: - """Calls GoodData Python SDK to retrieve all users. - - Returns: - list[CatalogUser]: A list of all users in the domain. - """ - return self._sdk.catalog_user.list_users() - - def create_or_update_user_group( - self, catalog_user_group: CatalogUserGroup, **_: str - ) -> None: - """Calls GoodData Python SDK to create or update a user group. - - Args: - catalog_user_group (CatalogUserGroup): The user group object to create or update. - Returns: - None - Raises: - GoodDataApiException: If the user group cannot be created or updated, - an exception is raised with additional context information. - """ - return self._sdk.catalog_user.create_or_update_user_group( - catalog_user_group - ) - - def delete_user_group(self, user_group_id: str) -> None: - """Calls GoodData Python SDK to delete a user group by its ID. - - Args: - user_group_id (str): The ID of the user group to delete. - Returns: - None - Raises: - GoodDataApiException: If the user group cannot be deleted, an exception - is raised with additional context information. - """ - return self._sdk.catalog_user.delete_user_group(user_group_id) - - def get_declarative_workspace_data_filters( - self, - ) -> CatalogDeclarativeWorkspaceDataFilters: - """Retrieves the declarative workspace data filters. - - Returns: - CatalogDeclarativeWorkspaceDataFilters: The declarative workspace data filters. - Raises: - GoodDataApiException: If the declarative workspace data filters cannot be retrieved, - an exception is raised with additional context information. - """ - return ( - self._sdk.catalog_workspace.get_declarative_workspace_data_filters() - ) - - def list_user_data_filters( - self, workspace_id: str - ) -> list[CatalogUserDataFilter]: - """Lists all user data filters for a given workspace. - - Args: - workspace_id (str): The ID of the workspace for which to list user data filters. - Returns: - list[CatalogUserDataFilter]: A list of user data filters for the specified workspace. - Raises: - GoodDataApiException: If the user data filters cannot be listed, an exception - is raised with additional context information. - """ - return self._sdk.catalog_workspace.list_user_data_filters(workspace_id) - - def delete_user_data_filter( - self, workspace_id: str, user_data_filter_id: str - ) -> None: - """Deletes a user data filter by its ID in the specified workspace. - - Args: - workspace_id (str): The ID of the workspace containing the user data filter. - user_data_filter_id (str): The ID of the user data filter to delete. - Returns: - None - Raises: - GoodDataApiException: If the user data filter cannot be deleted, an exception - is raised with additional context information. - """ - self._sdk.catalog_workspace.delete_user_data_filter( - workspace_id, user_data_filter_id - ) - - def create_or_update_user_data_filter( - self, workspace_id: str, user_data_filter: CatalogUserDataFilter - ) -> None: - """Creates or updates a user data filter in the specified workspace. - - Args: - workspace_id (str): The ID of the workspace where the user data filter - should be created or updated. - user_data_filter (CatalogUserDataFilter): The user data filter object to create or update. - Returns: - None - Raises: - GoodDataApiException: If the user data filter cannot be created or updated, - an exception is raised with additional context information. - """ - self._sdk.catalog_workspace.create_or_update_user_data_filter( - workspace_id, user_data_filter - ) - - def store_declarative_workspace( - self, workspace_id: str, export_path: Path - ) -> None: - """Stores the declarative workspace in the specified export path.""" - self._sdk.catalog_workspace.store_declarative_workspace( - workspace_id, export_path - ) - - def store_declarative_filter_views( - self, workspace_id: str, export_path: Path - ) -> None: - """Stores the declarative filter views in the specified export path.""" - self._sdk.catalog_workspace.store_declarative_filter_views( - workspace_id, export_path - ) diff --git a/packages/gooddata-pipelines/src/gooddata_pipelines/api/utils.py b/packages/gooddata-pipelines/src/gooddata_pipelines/api/utils.py deleted file mode 100644 index 10fc7a31f..000000000 --- a/packages/gooddata-pipelines/src/gooddata_pipelines/api/utils.py +++ /dev/null @@ -1,43 +0,0 @@ -# (C) 2025 GoodData Corporation - -"""Utility functions for GoodData Cloud API interactions.""" - -from typing import Any, Callable - -from gooddata_api_client import ApiException # type: ignore - -from gooddata_pipelines.api.exceptions import GoodDataApiException - - -def raise_with_context(**context_kwargs: str) -> Callable: - """ - Decorator to catch exceptions raised by SDK methods and raise a GoodDataApiException - with additional context information. - - Args: - context_kwargs (dict): Additional context information to include in the - GoodDataApiException. - """ - - def decorator(fn: Callable) -> Callable: - def wrapper(*method_args: Any, **method_kwargs: Any) -> Callable: - try: - return fn(*method_args, **method_kwargs) - except Exception as e: - # Process known exceptions - if isinstance(e, ApiException): - context_kwargs["http_status"] = f"{e.status} {e.reason}" - exception_content = e.body - else: - exception_content = str(e) - - # Format the exception message: "{exception_type}: {exception_content}" - message = f"{type(e).__name__}: {exception_content}" - - raise GoodDataApiException( - message, **context_kwargs, **method_kwargs - ) - - return wrapper - - return decorator