Coverage for rdflib_ocdm/counter_handler/filesystem_counter_handler.py: 100%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2016, Silvio Peroni <essepuntato@gmail.com> 

4# 

5# Permission to use, copy, modify, and/or distribute this software for any purpose 

6# with or without fee is hereby granted, provided that the above copyright notice 

7# and this permission notice appear in all copies. 

8# 

9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

15# SOFTWARE. 

16from __future__ import annotations 

17 

18import json 

19import os 

20 

21from rdflib_ocdm.counter_handler.counter_handler import CounterHandler 

22from rdflib_ocdm.support import is_string_empty 

23 

24 

25class FilesystemCounterHandler(CounterHandler): 

26 """A concrete implementation of the ``CounterHandler`` interface that persistently stores 

27 the counter values within the filesystem.""" 

28 

29 def __init__(self, info_dir: str) -> None: 

30 """ 

31 Constructor of the ``FilesystemCounterHandler`` class. 

32 

33 :param info_dir: The path to the folder that does/will contain the counter values. 

34 :type info_dir: str 

35 :raises ValueError: if ``info_dir`` is None or an empty string. 

36 """ 

37 if info_dir is None or is_string_empty(info_dir): 

38 raise ValueError("info_dir parameter is required!") 

39 

40 if info_dir[-1] != os.sep: 

41 info_dir += os.sep 

42 

43 self.info_dir: str = info_dir 

44 self.prov_files = dict() 

45 self.provenance_index_filename = 'provenance_index.json' 

46 

47 def set_counter(self, new_value: int, entity_name: str) -> None: 

48 """ 

49 It allows to set the counter value of provenance entities. 

50 

51 :param new_value: The new counter value to be set 

52 :type new_value: int 

53 :param entity_name: The entity name 

54 :type entity_name: str 

55 :raises ValueError: if ``new_value`` is a negative integer. 

56 :return: None 

57 """ 

58 entity_name = str(entity_name) 

59 if new_value < 0: 

60 raise ValueError("new_value must be a non negative integer!") 

61 file_path: str = self._get_prov_path() 

62 self.__initialize_file_if_not_existing(file_path, entity_name) 

63 with open(file_path, 'r', encoding='utf8') as file: 

64 data = json.load(file) 

65 with open(file_path, 'w', encoding='utf8') as outfile: 

66 data[entity_name] = new_value 

67 json.dump(obj=data, fp=outfile, ensure_ascii=False, indent=False) 

68 

69 def read_counter(self, entity_name: str) -> int: 

70 """ 

71 It allows to read the counter value of provenance entities. 

72 

73 :param entity_name: The entity name 

74 :type entity_name: str 

75 :return: The requested counter value. 

76 """ 

77 entity_name = str(entity_name) 

78 file_path: str = self._get_prov_path() 

79 return self._read_number(file_path, entity_name) 

80 

81 def increment_counter(self, entity_name: str) -> int: 

82 """ 

83 It allows to increment the counter value of provenance entities by one unit. 

84 

85 :param entity_name: The entity name 

86 :type entity_name: str 

87 :return: The newly-updated (already incremented) counter value. 

88 """ 

89 entity_name = str(entity_name) 

90 file_path: str = self._get_prov_path() 

91 return self._add_number(file_path, entity_name) 

92 

93 def _get_prov_path(self) -> str: 

94 return os.path.join(self.info_dir, self.provenance_index_filename) 

95 

96 def __initialize_file_if_not_existing(self, file_path: str, entity_name: str): 

97 entity_name = str(entity_name) 

98 if not os.path.exists(os.path.dirname(file_path)): 

99 os.makedirs(os.path.dirname(file_path)) 

100 if not os.path.isfile(file_path): 

101 with open(file_path, 'w', encoding='utf8') as outfile: 

102 json.dump({entity_name: 0}, ensure_ascii=False, indent=None, fp=outfile) 

103 

104 def _read_number(self, file_path: str, entity_name: str) -> int: 

105 self.__initialize_file_if_not_existing(file_path, entity_name) 

106 with open(file_path, 'r', encoding='utf8') as file: 

107 data = json.load(file) 

108 if entity_name in data: 

109 self.prov_files[entity_name] = data[entity_name] 

110 else: 

111 self.prov_files[entity_name] = 0 

112 return self.prov_files[entity_name] 

113 

114 def _add_number(self, file_path: str, entity_name: str) -> int: 

115 self.__initialize_file_if_not_existing(file_path, entity_name) 

116 cur_number = self._read_number(file_path, entity_name) 

117 cur_number += 1 

118 with open(file_path, 'r', encoding='utf8') as file: 

119 data = json.load(file) 

120 with open(file_path, 'w', encoding='utf8') as outfile: 

121 data[entity_name] = cur_number 

122 json_object = json.dumps(data, ensure_ascii=False, indent=None) 

123 outfile.write(json_object) 

124 return cur_number