115 lines
2.6 KiB
Python
115 lines
2.6 KiB
Python
import yaml
|
|
from pathlib import Path
|
|
from specs import Spec
|
|
from typing import List, Dict
|
|
|
|
|
|
class InvalidKeyError(Exception):
|
|
def __init__(self, key):
|
|
message = "Invalid key: {}".format(key)
|
|
|
|
super().__init__(key)
|
|
|
|
|
|
class MissingKeyError(Exception):
|
|
def __init__(self, key):
|
|
message = "Missing key: {}".format(key)
|
|
|
|
super().__init__(key)
|
|
|
|
|
|
def parse_specs_file(path: Path) -> List[Spec]:
|
|
"""
|
|
Parse a YAML file defining backup specs.
|
|
|
|
Args:
|
|
path: path to the specs file
|
|
|
|
Returns:
|
|
A list of specs
|
|
"""
|
|
|
|
# Skeleton of a spec config
|
|
# If a value is None, this means it doesn't have a default value and must be
|
|
# defined
|
|
spec_skel = {
|
|
"source": None,
|
|
"destination": None,
|
|
"limit": None,
|
|
"volume": False,
|
|
"notify": {
|
|
"title": "Backup Notification",
|
|
"events": ["failure"]
|
|
}
|
|
}
|
|
|
|
# Read YAML file
|
|
with open(path, "r") as yaml_file:
|
|
data = yaml.load(yaml_file, Loader=yaml.Loader)
|
|
|
|
# Check specs section exists
|
|
if "specs" not in data:
|
|
raise MissingKeyError("specs")
|
|
|
|
# Allow for default notify settings
|
|
if "notify" in data:
|
|
spec_skel["notify"] = data["notify"]
|
|
|
|
specs = []
|
|
# Check format for each spec
|
|
for key in data["specs"]:
|
|
specs.append(Spec.from_dict(key, combine_with_skeleton(
|
|
data["specs"][key], spec_skel)
|
|
))
|
|
|
|
return specs
|
|
|
|
|
|
def combine_with_skeleton(data: Dict, skel: Dict) -> Dict:
|
|
"""
|
|
Compare a dict with a given skeleton dict, and fill in default values where
|
|
needed.
|
|
"""
|
|
|
|
# First, check for illegal keys
|
|
for key in data:
|
|
if key not in skel:
|
|
raise InvalidKeyError(key)
|
|
|
|
# Then, check the default values
|
|
for key, value in skel.items():
|
|
if key not in data:
|
|
# Raise error if there's not default value
|
|
if value is None:
|
|
raise MissingKeyError(key)
|
|
|
|
# Replace with default value
|
|
data[key] = value
|
|
|
|
# Error if value is not same type as default value
|
|
elif type(data[key]) != type(value) and value is not None:
|
|
raise TypeError("Invalid value type")
|
|
|
|
# Recurse into dicts
|
|
elif type(value) == dict:
|
|
data[key] = combine_with_skeleton(data[key], value)
|
|
|
|
return data
|
|
|
|
|
|
# Test cases
|
|
if __name__ == "__main__":
|
|
d1 = {
|
|
"a": 5
|
|
}
|
|
s1 = {
|
|
"a": 7,
|
|
"b": 2
|
|
}
|
|
r1 = {
|
|
"a": 5,
|
|
"b": 2
|
|
}
|
|
|
|
assert combine_with_skeleton(d1, s1) == r1
|