Coverage for src/lektor_ng/databags.py: 81%

1import errno 

2import json 

3import os 

4from collections import OrderedDict 

5 

6from lektor_ng.context import get_ctx 

7from lektor_ng.inifile import IniFile 

8from lektor_ng.utils import ( 

9 decode_flat_data, 

10 iter_dotted_path_prefixes, 

11 merge, 

12 resolve_dotted_value, 

13) 

14 

15 

16def load_databag(filename): 

17 try: 

18 if filename.endswith(".json"): 

19 with open(filename, encoding="utf-8") as f: 

20 return json.load(f, object_pairs_hook=OrderedDict) 

21 elif filename.endswith(".ini"): 

22 return decode_flat_data(IniFile(filename).items(), dict_cls=OrderedDict) 

23 else: 

24 return None 

25 except OSError as e: 

26 if e.errno != errno.ENOENT: 

27 raise 

28 return None 

29 

30 

31class Databags: 

32 def __init__(self, env): 

33 self.env = env 

34 self.root_path = os.path.join(self.env.root_path, "databags") 

35 self._known_bags = {} 

36 self._bags = {} 

37 try: 

38 for filename in os.listdir(self.root_path): 

39 if filename.endswith((".ini", ".json")): 

40 self._known_bags.setdefault(filename.rsplit(".", -1)[0], []).append(filename) 

41 except OSError: 

42 pass 

43 

44 def get_bag(self, name): 

45 sources = self._known_bags.get(name) 

46 if not sources: 

47 return None 

48 rv = self._bags.get(name) 

49 if rv is None: 

50 filenames = [] 

51 rv = OrderedDict() 

52 for filename in sources: 

53 filename = os.path.join(self.root_path, filename) 

54 rv = merge(rv, load_databag(filename)) 

55 filenames.append(filename) 

56 self._bags[name] = (rv, filenames) 

57 else: 

58 rv, filenames = rv 

59 

60 ctx = get_ctx() 

61 if ctx is not None: 

62 for filename in filenames: 

63 ctx.record_dependency(filename) 

64 

65 return rv 

66 

67 def lookup(self, key): 

68 for prefix, local_key in iter_dotted_path_prefixes(key): 

69 bag = self.get_bag(prefix) 

70 if bag is not None: 

71 if local_key is None: 

72 return bag 

73 return resolve_dotted_value(bag, local_key) 

74 return None