Coverage for b4_backup/config_schema.py: 100%

49 statements  

« prev     ^ index     » next       coverage.py v7.9.1, created at 2025-06-18 22:40 +0000

1""" 

2This module contains the config structure and it's default values. 

3 

4The config is using the YAML syntax and this file describes the structure of it. 

5""" 

6 

7import textwrap 

8from dataclasses import dataclass, field 

9from enum import Enum 

10from pathlib import Path, PurePath 

11from typing import Any 

12 

13import omegaconf 

14from omegaconf import II 

15 

16DEFAULT = "_default" 

17 

18 

19class TargetRestoreStrategy(str, Enum): 

20 """ 

21 Specifies the restore procedure to be used. 

22 

23 Attributes: 

24 SAFE: Just copy the snapshot back to the source snapshot directory without touching the target directory 

25 REPLACE: Bases safe, but also replace the target subvolumes with the copied one. Works by moving the original target away and then copy the snapshot to that place. Revertable by using the REPLACE snapshot name. 

26 """ 

27 

28 SAFE = "safe" 

29 REPLACE = "replace" 

30 

31 

32class SubvolumeBackupStrategy(str, Enum): 

33 """ 

34 Backup strategy for subvolumes. 

35 

36 Attributes: 

37 IGNORE: The subvolume will be ignored during the backup 

38 SOURCE_ONLY: The subvolume will be kept only on source and not sent to destination 

39 FULL: The subvolume will be sent to destination 

40 """ 

41 

42 IGNORE = "ignore" 

43 SOURCE_ONLY = "source_only" 

44 FULL = "full" 

45 

46 

47class SubvolumeFallbackStrategy(str, Enum): 

48 """ 

49 Fallback strategy for subvolumes on a restore if the backup subvolume is already deleted. 

50 

51 Attributes: 

52 DROP: The subvolume is lost after a restore (Use case: Docker artifacts or everywhere else where btrfs subvolumes are created dynamically) 

53 NEW: An empty subvolume is created at that place (Use case: Cache directories) 

54 KEEP: The old subvolume will be copied at the new place, if doesn't exist, a new one will be created (Use case: Steam library) 

55 """ 

56 

57 DROP = "drop" 

58 NEW = "new" 

59 KEEP = "keep" 

60 

61 

62class OnDestinationDirNotFound(str, Enum): 

63 """ 

64 How to behave, if the destination directory does not exist. 

65 

66 Attributes: 

67 CREATE: Create the missing directory structure and proceed without an error. 

68 FAIL: Throw an error and stop execution. 

69 """ 

70 

71 CREATE = "create" 

72 FAIL = "fail" 

73 

74 

75@dataclass 

76class TargetSubvolume: 

77 """ 

78 Defines how to handle a specific subvolume in a target. 

79 

80 Args: 

81 backup_strategy: How to handle the subvolume during backup 

82 fallback_strategy: How to handle the subvolume during restore if the backup subvolume is already deleted 

83 """ 

84 

85 backup_strategy: SubvolumeBackupStrategy = II(f"..{DEFAULT}.backup_strategy") 

86 fallback_strategy: SubvolumeFallbackStrategy = II(f"..{DEFAULT}.fallback_strategy") 

87 

88 

89@dataclass 

90class BackupTarget: 

91 """ 

92 Defines a single backup target. 

93 

94 Args: 

95 source: Path or URL you want to backup. Needs to be a btrfs subvolume 

96 destination: Path or URL where you want to send snapshots. If None, snapshots will only be on source side 

97 restore_strategy: Default procedure to restore a backup 

98 src_snapshot_dir: Directory where source snapshots relative to the mount point of the btrfs volume are located 

99 src_retention: Retention rules for snapshots located at the source 

100 dst_retention: Retention rules for snapshots located at the destination 

101 replaced_target_ttl: The minimum time the old replaced subvolume should be kept 

102 subvolume_rules: Contains rules for how to handle the subvolumes of a target 

103 """ 

104 

105 source: str | None = II(f"..{DEFAULT}.source") 

106 destination: str | None = II(f"..{DEFAULT}.destination") 

107 if_dst_dir_not_found: OnDestinationDirNotFound = II(f"..{DEFAULT}.if_dst_dir_not_found") 

108 restore_strategy: TargetRestoreStrategy = II(f"..{DEFAULT}.restore_strategy") 

109 src_snapshot_dir: Path = II(f"..{DEFAULT}.src_snapshot_dir") 

110 src_retention: dict[str, dict[str, str]] = field(default_factory=dict) 

111 dst_retention: dict[str, dict[str, str]] = field(default_factory=dict) 

112 replaced_target_ttl: str = II(f"..{DEFAULT}.replaced_target_ttl") 

113 subvolume_rules: dict[str, TargetSubvolume] = II(f"..{DEFAULT}.subvolume_rules") 

114 

115 

116@dataclass 

117class BaseConfig: 

118 """ 

119 The root level of the configuration. 

120 

121 Args: 

122 backup_targets: An object containing all targets to backup 

123 default_targets: List of default targets to use if not specified 

124 timezone: Timezone to use 

125 logging: Python logging configuration settings (logging.config.dictConfig). 

126 """ 

127 

128 backup_targets: dict[str, BackupTarget] = field( 

129 default_factory=lambda: { 

130 DEFAULT: BackupTarget( 

131 source=None, 

132 destination=None, 

133 if_dst_dir_not_found=OnDestinationDirNotFound.CREATE, 

134 restore_strategy=TargetRestoreStrategy.SAFE, 

135 src_snapshot_dir=Path(".b4_backup"), 

136 src_retention={DEFAULT: {"all": "1"}}, 

137 dst_retention={DEFAULT: {"all": "forever"}}, 

138 replaced_target_ttl="24hours", 

139 subvolume_rules={ 

140 DEFAULT: TargetSubvolume( 

141 backup_strategy=SubvolumeBackupStrategy.FULL, 

142 fallback_strategy=SubvolumeFallbackStrategy.DROP, 

143 ), 

144 "/": TargetSubvolume(), 

145 }, 

146 ) 

147 } 

148 ) 

149 

150 default_targets: list[str] = field(default_factory=list) 

151 timezone: str = "utc" 

152 

153 logging: dict[str, Any] = II( 

154 "oc.create:${from_file:" + str(Path(__file__).parent / "default_logging_config.yml") + "}" 

155 ) 

156 

157 # Internal only 

158 # May change at runtime 

159 config_path: Path = field(default=Path("~/.config/b4_backup.yml")) 

160 

161 def __post_init__(self): 

162 """Used for validation of the values.""" 

163 options = set(self.backup_targets) - {DEFAULT} 

164 for target in self.default_targets: 

165 if not any(PurePath(x).is_relative_to(target) for x in options): 

166 raise omegaconf.errors.ValidationError( 

167 textwrap.dedent( 

168 f"""\ 

169 Item '{target}' is not in 'backup_targets' but defined in 'default_targets' 

170 full_key: default_targets 

171 object_type={self.__class__.__name__}""" 

172 ) 

173 )