feat(gooddata-pipelines): add workspace restore

Author: janmatzek

gooddata-pipelines/README.md

@@ -10,9 +10,10 @@ You can use the package to manage following resources in GDC:
    - User/Group permissions
    - User Data Filters
    - Child workspaces (incl. Workspace Data Filter settings)
-1. _[PLANNED]:_ Backup and restore of workspaces
-1. _[PLANNED]:_ Custom fields management
-   - extend the Logical Data Model of a child workspace
+1. Backup and restore of workspaces
+   - Create and backup snapshots of workspace metadata.
+1. LDM Extension
+   - extend the Logical Data Model of a child workspace with custom datasets and fields
 
 In case you are not interested in incorporating a library in your own program but would like to use a ready-made script, consider having a look at [GoodData Productivity Tools](https://github.com/gooddata/gooddata-productivity-tools).
 

gooddata-pipelines/gooddata_pipelines/__init__.py

@@ -10,6 +10,10 @@
     S3StorageConfig,
     StorageType,
 )
+from .backup_and_restore.restore_manager import (
+    RestoreManager,
+    WorkspaceToRestore,
+)
 from .backup_and_restore.storage.local_storage import LocalStorage
 from .backup_and_restore.storage.s3_storage import S3Storage
 
@@ -57,6 +61,8 @@
 
 __all__ = [
     "BackupManager",
+    "RestoreManager",
+    "WorkspaceToRestore",
     "BackupRestoreConfig",
     "StorageType",
     "LocalStorage",

gooddata-pipelines/gooddata_pipelines/api/gooddata_api.py

@@ -167,13 +167,40 @@ def get_user_data_filters(self, workspace_id: str) -> requests.Response:
         endpoint = f"/layout/workspaces/{workspace_id}/userDataFilters"
         return self._get(endpoint)
 
+    def put_user_data_filters(
+        self, workspace_id: str, user_data_filters: dict[str, Any]
+    ) -> requests.Response:
+        """Puts the user data filters into GoodData workspace."""
+        headers = {**self.headers, "Content-Type": "application/json"}
+        return self._put(
+            f"/layout/workspaces/{workspace_id}/userDataFilters",
+            user_data_filters,
+            headers,
+        )
+
     def get_automations(self, workspace_id: str) -> requests.Response:
         """Gets the automations for a given workspace."""
         endpoint = (
             f"/entities/workspaces/{workspace_id}/automations?include=ALL"
         )
         return self._get(endpoint)
 
+    def post_automation(
+        self, workspace_id: str, automation: dict[str, Any]
+    ) -> requests.Response:
+        """Posts an automation for a given workspace."""
+        endpoint = f"/entities/workspaces/{workspace_id}/automations"
+        return self._post(endpoint, automation)
+
+    def delete_automation(
+        self, workspace_id: str, automation_id: str
+    ) -> requests.Response:
+        """Deletes an automation for a given workspace."""
+        endpoint = (
+            f"/entities/workspaces/{workspace_id}/automations/{automation_id}"
+        )
+        return self._delete(endpoint)
+
     def get_all_metrics(self, workspace_id: str) -> requests.Response:
         """Get all metrics from the specified workspace.
 

gooddata-pipelines/gooddata_pipelines/backup_and_restore/backup_manager.py

@@ -1,42 +1,31 @@
 # (C) 2025 GoodData Corporation
 
-import json
 import os
 import shutil
 import tempfile
 import time
 import traceback
 from pathlib import Path
-from typing import Any, Type
+from typing import Any
 
 import attrs
 import requests
-import yaml
-from gooddata_sdk.utils import PROFILES_FILE_PATH, profile_content
 
-from gooddata_pipelines.api.gooddata_api_wrapper import GoodDataApi
 from gooddata_pipelines.backup_and_restore.backup_input_processor import (
     BackupInputProcessor,
 )
+from gooddata_pipelines.backup_and_restore.base_manager import BaseManager
 from gooddata_pipelines.backup_and_restore.constants import (
     BackupSettings,
     DirNames,
 )
 from gooddata_pipelines.backup_and_restore.models.input_type import InputType
 from gooddata_pipelines.backup_and_restore.models.storage import (
     BackupRestoreConfig,
-    StorageType,
 )
 from gooddata_pipelines.backup_and_restore.storage.base_storage import (
     BackupStorage,
 )
-from gooddata_pipelines.backup_and_restore.storage.local_storage import (
-    LocalStorage,
-)
-from gooddata_pipelines.backup_and_restore.storage.s3_storage import (
-    S3Storage,
-)
-from gooddata_pipelines.logger import LogObserver
 from gooddata_pipelines.utils.rate_limiter import RateLimiter
 
 
@@ -45,16 +34,12 @@ class BackupBatch:
     list_of_ids: list[str]
 
 
-class BackupManager:
+class BackupManager(BaseManager):
     storage: BackupStorage
 
     def __init__(self, host: str, token: str, config: BackupRestoreConfig):
-        self._api = GoodDataApi(host, token)
-        self.logger = LogObserver()
-
-        self.config = config
+        super().__init__(host, token, config)
 
-        self.storage = self._get_storage(self.config)
         self.org_id = self._api.get_organization_id()
 
         self.loader = BackupInputProcessor(self._api, self.config.api_page_size)
@@ -63,39 +48,6 @@ def __init__(self, host: str, token: str, config: BackupRestoreConfig):
             calls_per_second=self.config.api_calls_per_second,
         )
 
-    @classmethod
-    def create(
-        cls: Type["BackupManager"],
-        config: BackupRestoreConfig,
-        host: str,
-        token: str,
-    ) -> "BackupManager":
-        """Creates a backup worker instance using the provided host and token."""
-        return cls(host=host, token=token, config=config)
-
-    @classmethod
-    def create_from_profile(
-        cls: Type["BackupManager"],
-        config: BackupRestoreConfig,
-        profile: str = "default",
-        profiles_path: Path = PROFILES_FILE_PATH,
-    ) -> "BackupManager":
-        """Creates a backup worker instance using a GoodData profile file."""
-        content = profile_content(profile, profiles_path)
-        return cls(**content, config=config)
-
-    @staticmethod
-    def _get_storage(conf: BackupRestoreConfig) -> BackupStorage:
-        """Returns the storage class based on the storage type."""
-        if conf.storage_type == StorageType.S3:
-            return S3Storage(conf)
-        elif conf.storage_type == StorageType.LOCAL:
-            return LocalStorage(conf)
-        else:
-            raise RuntimeError(
-                f'Unsupported storage type "{conf.storage_type.value}".'
-            )
-
     def get_user_data_filters(self, ws_id: str) -> dict:
         """Returns the user data filters for the specified workspace."""
         with self._api_rate_limiter:
@@ -133,19 +85,13 @@ def _store_user_data_filters(
                 "user_data_filters",
                 filter["id"] + ".yaml",
             )
-            self._write_to_yaml(udf_file_path, filter)
+            self.yaml_utils.dump(udf_file_path, filter)
 
     @staticmethod
     def _move_folder(source: Path, destination: Path) -> None:
         """Moves the source folder to the destination."""
         shutil.move(source, destination)
 
-    @staticmethod
-    def _write_to_yaml(path: str, source: Any) -> None:
-        """Writes the source to a YAML file."""
-        with open(path, "w") as outfile:
-            yaml.dump(source, outfile)
-
     def _get_automations_from_api(self, workspace_id: str) -> Any:
         """Returns automations for the workspace as JSON."""
         with self._api_rate_limiter:
@@ -182,8 +128,7 @@ def _store_automations(self, export_path: Path, workspace_id: str) -> None:
 
         # Store the automations in a JSON file
         if len(automations["data"]) > 0:
-            with open(automations_file_path, "w") as f:
-                json.dump(automations, f)
+            self.json_utils.dump(automations_file_path, automations)
 
     def store_declarative_filter_views(
         self, export_path: Path, workspace_id: str

gooddata-pipelines/gooddata_pipelines/backup_and_restore/base_manager.py

@@ -0,0 +1,73 @@
+# (C) 2025 GoodData Corporation
+
+import abc
+from pathlib import Path
+from typing import Type, TypeVar
+
+from gooddata_sdk.utils import PROFILES_FILE_PATH, profile_content
+
+from gooddata_pipelines.api.gooddata_api_wrapper import GoodDataApi
+from gooddata_pipelines.backup_and_restore.models.storage import (
+    BackupRestoreConfig,
+    StorageType,
+)
+from gooddata_pipelines.backup_and_restore.storage.base_storage import (
+    BackupStorage,
+)
+from gooddata_pipelines.backup_and_restore.storage.local_storage import (
+    LocalStorage,
+)
+from gooddata_pipelines.backup_and_restore.storage.s3_storage import S3Storage
+from gooddata_pipelines.logger import LogObserver
+from gooddata_pipelines.utils.file_utils import JsonUtils, YamlUtils
+
+ManagerT = TypeVar("ManagerT", bound="BaseManager")
+
+
+class BaseManager(abc.ABC):
+    """Base class to provide constructors for backup and restore managers."""
+
+    storage: BackupStorage
+
+    def __init__(self, host: str, token: str, config: BackupRestoreConfig):
+        self.config = config
+
+        self._api: GoodDataApi = GoodDataApi(host, token)
+        self.logger: LogObserver = LogObserver()
+
+        self.storage = self._get_storage(self.config)
+
+        self.yaml_utils = YamlUtils()
+        self.json_utils = JsonUtils()
+
+    def _get_storage(self, conf: BackupRestoreConfig) -> BackupStorage:
+        """Returns the storage class based on the storage type."""
+        if conf.storage_type == StorageType.S3:
+            return S3Storage(conf)
+        elif conf.storage_type == StorageType.LOCAL:
+            return LocalStorage(conf)
+        else:
+            raise RuntimeError(
+                f'Unsupported storage type "{conf.storage_type.value}".'
+            )
+
+    @classmethod
+    def create(
+        cls: Type[ManagerT],
+        config: BackupRestoreConfig,
+        host: str,
+        token: str,
+    ) -> ManagerT:
+        """Creates a backup worker instance using the provided host and token."""
+        return cls(host=host, token=token, config=config)
+
+    @classmethod
+    def create_from_profile(
+        cls: Type[ManagerT],
+        config: BackupRestoreConfig,
+        profile: str = "default",
+        profiles_path: Path = PROFILES_FILE_PATH,
+    ) -> ManagerT:
+        """Creates a backup worker instance using a GoodData profile file."""
+        content = profile_content(profile, profiles_path)
+        return cls(host=content["host"], token=content["token"], config=config)

gooddata-pipelines/gooddata_pipelines/backup_and_restore/constants.py

@@ -1,12 +1,12 @@
 # (C) 2025 GoodData Corporation
 import datetime
+from enum import Enum
 
 import attrs
 from gooddata_sdk._version import __version__ as sdk_version
 
 
-@attrs.frozen
-class DirNames:
+class DirNames(str, Enum):
     """
     Folder names used in the SDK backup process:
     - LAYOUTS - GoodData Layouts
@@ -23,13 +23,14 @@ class DirNames:
 
 @attrs.frozen
 class ApiDefaults:
-    DEFAULT_PAGE_SIZE = 100
-    DEFAULT_BATCH_SIZE = 100
-    DEFAULT_API_CALLS_PER_SECOND = 1.0
+    PAGE_SIZE = 100
+    BATCH_SIZE = 100
+    CALLS_PER_SECOND = 1.0
 
 
 @attrs.frozen
-class BackupSettings(ApiDefaults):
+class BackupSettings:
+    API = ApiDefaults()
     MAX_RETRIES = 3
     RETRY_DELAY = 5  # seconds
     TIMESTAMP_SDK_FOLDER = (