26 add a feature that enables defining working directories

README.md

@@ -91,18 +91,27 @@ These represent their matching restic commands: [restic backup][restic-docs-back
 #### Backup
 
 With the backup step a path can be backed up to a repository.
-The `backup` step needs to have the following fields:
-- `repository`: This is a reference to a repository via the repository id
-- `source_path`: The path that needs to be backed up to the repository
 
-Optionally the `tags` field can be provided with a list of tags that needs to be applied to the new snapshot.
+The `backup` step needs to have the following fields:
+- `repository`: This is a reference to a repository via the repository id. Must be a string.
+- `source_path`: The path that needs to be backed up to the repository.
+  If a `working_dir` is defined, then it must be a relative path,
+  otherwise it must be an absolute path. Must be a string.
+
+The following fields are optional:
+- `tags`: list of tags that needs to be applied to the new snapshot
+- `working_dir`: The restic command will be executed in this directory.
+  The working directory will be changed only temporarily while executing the step
+  and after execution the previous value will be restored.
+  If undefined, no change will be done. Must be a string.
 
 #### Copy
 
 With the copy step snapshots can be copied between repositories.
+
 The `copy` step needs to have the following fields:
-- `source_repository`: Reference to the soruce repository via the repository id
-- `target_repository`: Reference to the target repository via the repository id
+- `source_repository`: Reference to the source repository via the repository id. Must be a string.
+- `target_repository`: Reference to the target repository via the repository id. Must be a string.
 
 ### Simple example
 

pyproject.toml

@@ -14,10 +14,10 @@ dynamic = ["version"]
 keywords = [
     "restic",
     "backup",
-    "automation"]
+    "automation"
+]
 classifiers = [
     "Programming Language :: Python :: 3.14",
-    "Private :: Do Not Upload"
 ]
 
 [project.optional-dependencies]

source/backup_automation/__init__.py

@@ -1 +1 @@
-__version__ = "0.3.0"
+__version__ = "0.4.0"

source/backup_automation/json_config.py

@@ -0,0 +1,42 @@
+from typing import Type, TypeVar
+
+from backup_automation.typehints import JsonDict, JsonDictKey
+
+
+T = TypeVar("T")
+
+
+class JsonConfigException(Exception):
+    """
+    Exception to raise when restic playbook related errors happen.
+    """
+
+
+def __get_config_value(configuration: JsonDict, key: JsonDictKey, expected_type: Type[T]) -> T:
+    value = configuration[key]
+    if not isinstance(value, expected_type):
+        raise JsonConfigException(f"Expected type {expected_type} for key {key} but got {type(value)} instead.")
+
+    return value
+
+
+def get_optional_config_value(configuration: JsonDict, key: JsonDictKey, expected_type: Type[T]) -> T | None:
+    """
+    Returns value from configuration if key is present, returns None otherwise.
+    If the key is present but the value is not of the expected type, raises JsonConfigException.
+    """
+    if key not in configuration:
+        return None
+
+    return __get_config_value(configuration, key, expected_type)
+
+
+def get_required_config_value(configuration: JsonDict, key: JsonDictKey, expected_type: Type[T]) -> T:
+    """
+    Returns value from configuration if key is present, raises JsonConfigException otherwise.
+    If the key is present but the value is not of the expected type, raises JsonConfigException.
+    """
+    if key not in configuration:
+        raise JsonConfigException(f"Missing required key in configuration: {key}")
+
+    return __get_config_value(configuration, key, expected_type)

source/backup_automation/restic/restic_playbook_format.py

@@ -22,5 +22,6 @@ class ResticPlaybookFormat(PlaybookFormat):
     STEPS_BACKUP_REPOSITORY_KEY = "repository"
     STEPS_BACKUP_SOURCE_PATH_KEY = "source_path"
     STEPS_BACKUP_TAGS_KEY = "tags"
+    STEPS_BACKUP_WORKING_DIR_KEY = "working_dir"
     STEPS_COPY_SOURCE_REPOSITORY_KEY = "source_repository"
     STEPS_COPY_TARGET_REPOSITORY_KEY = "target_repository"

source/backup_automation/restic/restic_playbook_step_parser.py

@@ -1,6 +1,7 @@
 import pathlib
 from typing import Callable
 
+from backup_automation.json_config import JsonConfigException, get_optional_config_value, get_required_config_value
 from backup_automation.restic.restic_backend import ResticBackend
 from backup_automation.restic.restic_playbook_exception import ResticPlaybookException
 from backup_automation.restic.restic_playbook_format import ResticPlaybookFormat
@@ -27,35 +28,59 @@ def parse(self, step_json: JsonDict) -> ResticPlaybookStep:
         """
         Parses a playbook step from received JSON object into a ResticPlaybookStep object.
         """
-        step_command = step_json[self.__format.STEPS_COMMAND_KEY]
-        match step_command:
-            case self.__format.STEPS_COMMAND_VALUE_BACKUP:
-                return self.__parse_backup_step(step_json)
-            case self.__format.STEPS_COMMAND_VALUE_COPY:
-                return self.__parse_copy_step(step_json)
-            case _:
-                raise ResticPlaybookException(f"Unexpected command in step: {step_json}")
+        try:
+            step_command = step_json[self.__format.STEPS_COMMAND_KEY]
+            match step_command:
+                case self.__format.STEPS_COMMAND_VALUE_BACKUP:
+                    return self.__parse_backup_step(step_json)
+                case self.__format.STEPS_COMMAND_VALUE_COPY:
+                    return self.__parse_copy_step(step_json)
+                case _:
+                    raise ResticPlaybookException(f"Unexpected command in step: {step_json}")
+        except (ResticPlaybookException, JsonConfigException) as e:
+            raise ResticPlaybookException(f"Could not parse playbook step: {step_json}") from e
 
     def __parse_backup_step(self, step_json: JsonDict) -> ResticPlaybookBackupStep:
-        repository_id = step_json[self.__format.STEPS_BACKUP_REPOSITORY_KEY]
+        repository_id = get_required_config_value(step_json, self.__format.STEPS_BACKUP_REPOSITORY_KEY, str)
         repository = self.__repository_lookup(repository_id)
 
-        source_path = pathlib.Path(step_json[self.__format.STEPS_BACKUP_SOURCE_PATH_KEY])
-        if not source_path.is_dir():
-            raise ResticPlaybookException(f"Source path is not a valid directory: {source_path}")
+        raw_source_path = get_required_config_value(step_json, self.__format.STEPS_BACKUP_SOURCE_PATH_KEY, str)
+        source_path = pathlib.Path(raw_source_path)
 
         tags = step_json.get(self.__format.STEPS_BACKUP_TAGS_KEY, [])
         if not isinstance(tags, list):
             raise ResticPlaybookException(f"Tags is not a valid JSON array: {tags}")
 
-        return ResticPlaybookBackupStep(self.__backend, repository, source_path, tuple(tags))
+        raw_working_dir = get_optional_config_value(step_json, self.__format.STEPS_BACKUP_WORKING_DIR_KEY, str)
+        working_dir = pathlib.Path(raw_working_dir) if raw_working_dir else None
+
+        if working_dir:
+            if source_path.is_absolute():
+                raise ResticPlaybookException(f"If working_dir is defined, source_path must be a relative path!"
+                                              f" working_dir: {working_dir} source_path: {source_path}")
+
+            path_to_backup = working_dir / source_path
+            if not path_to_backup.is_dir():
+                raise ResticPlaybookException(f"The path to backup (combination of working_dir and source_path)"
+                                              f" is not a valid directory: {path_to_backup}")
+        else:
+            if not source_path.is_absolute():
+                raise ResticPlaybookException(f"If working_dir is not defined, source_path must be an absolute path! "
+                                              f"source_path: {source_path}")
+
+            if not source_path.is_dir():
+                raise ResticPlaybookException(f"The path to backup (source_path)"
+                                              f" is not a valid directory: {source_path}")
+
+        return ResticPlaybookBackupStep(self.__backend, repository, source_path, tuple(tags), working_dir)
 
     def __parse_copy_step(self, step_json: JsonDict) -> ResticPlaybookCopyStep:
-        source_repository_id = step_json[self.__format.STEPS_COPY_SOURCE_REPOSITORY_KEY]
+        source_repository_id = get_required_config_value(step_json, self.__format.STEPS_COPY_SOURCE_REPOSITORY_KEY, str)
         source_repository = self.__repository_lookup(source_repository_id)
 
-        target_repository_id = step_json[self.__format.STEPS_COPY_TARGET_REPOSITORY_KEY]
+        target_repository_id = get_required_config_value(step_json, self.__format.STEPS_COPY_TARGET_REPOSITORY_KEY, str)
         target_repository = self.__repository_lookup(target_repository_id)
+
         if source_repository == target_repository:
             raise ResticPlaybookException("The source and target repositories cannot be the same!")
 

source/backup_automation/restic/restic_playbook_steps.py

@@ -1,3 +1,4 @@
+import contextlib
 import pathlib
 from abc import ABC
 
@@ -22,19 +23,26 @@ class ResticPlaybookBackupStep(ResticPlaybookStep):
     """
     Class to represent the restic specific backup playbook step.
     """
+    # pylint: disable=too-many-arguments
+    # pylint: disable=too-many-positional-arguments
+    # This class needs to represent a backup step with all its fields.
     def __init__(self,
                  backend: ResticBackend,
                  repository: ResticRepository,
                  source_path: pathlib.Path,
-                 tags: tuple[str, ...]):
+                 tags: tuple[str, ...],
+                 working_dir: pathlib.Path | None):
         super().__init__()
         self.__backend = backend
         self.__repository = repository
         self.__source_path = source_path
         self.__tags = tags
+        self.__working_dir = working_dir
 
     def execute(self) -> None:
-        self.__backend.backup(self.__repository, self.__source_path, self.__tags)
+        working_dir_context = contextlib.chdir(self.__working_dir) if self.__working_dir else contextlib.nullcontext()
+        with working_dir_context:
+            self.__backend.backup(self.__repository, self.__source_path, self.__tags)
 
 
 class ResticPlaybookCopyStep(ResticPlaybookStep):

source/backup_automation/typehints.py

@@ -1,5 +1,8 @@
 from typing import Any
 
 
-JsonDict = dict[str, Any]
-JsonList = list[Any]
+JsonDictKey = str
+JsonDictValue = Any
+JsonDict = dict[JsonDictKey, JsonDictValue]
+JsonListElement = Any
+JsonList = list[JsonListElement]