diff --git a/README.md b/README.md index 80c28a8..3148b68 100644 --- a/README.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 5aee45c..8eb442b 100644 --- a/pyproject.toml +++ b/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] diff --git a/source/backup_automation/__init__.py b/source/backup_automation/__init__.py index 493f741..6a9beea 100644 --- a/source/backup_automation/__init__.py +++ b/source/backup_automation/__init__.py @@ -1 +1 @@ -__version__ = "0.3.0" +__version__ = "0.4.0" diff --git a/source/backup_automation/json_config.py b/source/backup_automation/json_config.py new file mode 100644 index 0000000..6ab7f8e --- /dev/null +++ b/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) diff --git a/source/backup_automation/restic/restic_playbook_format.py b/source/backup_automation/restic/restic_playbook_format.py index 426c2ba..0ac9baa 100644 --- a/source/backup_automation/restic/restic_playbook_format.py +++ b/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" diff --git a/source/backup_automation/restic/restic_playbook_step_parser.py b/source/backup_automation/restic/restic_playbook_step_parser.py index e56f084..c7db764 100644 --- a/source/backup_automation/restic/restic_playbook_step_parser.py +++ b/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!") diff --git a/source/backup_automation/restic/restic_playbook_steps.py b/source/backup_automation/restic/restic_playbook_steps.py index 77840fa..c4422b9 100644 --- a/source/backup_automation/restic/restic_playbook_steps.py +++ b/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): diff --git a/source/backup_automation/typehints.py b/source/backup_automation/typehints.py index f754658..be7c1f8 100644 --- a/source/backup_automation/typehints.py +++ b/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]