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

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2016 Silvio Peroni <essepuntato@gmail.com> 

4# SPDX-FileCopyrightText: 2023 Arcangelo Massari <arcangelo.massari@unibo.it> 

5# 

6# SPDX-License-Identifier: ISC 

7from __future__ import annotations 

8 

9import json 

10import os 

11 

12from rdflib_ocdm.counter_handler.counter_handler import CounterHandler 

13from rdflib_ocdm.support import is_string_empty 

14 

15 

16class FilesystemCounterHandler(CounterHandler): 

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

18 the counter values within the filesystem.""" 

19 

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

21 """ 

22 Constructor of the ``FilesystemCounterHandler`` class. 

23 

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

25 :type info_dir: str 

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

27 """ 

28 if info_dir is None or is_string_empty(info_dir): 

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

30 

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

32 info_dir += os.sep 

33 

34 self.info_dir: str = info_dir 

35 self.prov_files = dict() 

36 self.provenance_index_filename = 'provenance_index.json' 

37 

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

39 """ 

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

41 

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

43 :type new_value: int 

44 :param entity_name: The entity name 

45 :type entity_name: str 

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

47 :return: None 

48 """ 

49 entity_name = str(entity_name) 

50 if new_value < 0: 

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

52 file_path: str = self._get_prov_path() 

53 self.__initialize_file_if_not_existing(file_path, entity_name) 

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

55 data = json.load(file) 

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

57 data[entity_name] = new_value 

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

59 

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

61 """ 

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

63 

64 :param entity_name: The entity name 

65 :type entity_name: str 

66 :return: The requested counter value. 

67 """ 

68 entity_name = str(entity_name) 

69 file_path: str = self._get_prov_path() 

70 return self._read_number(file_path, entity_name) 

71 

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

73 """ 

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

75 

76 :param entity_name: The entity name 

77 :type entity_name: str 

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

79 """ 

80 entity_name = str(entity_name) 

81 file_path: str = self._get_prov_path() 

82 return self._add_number(file_path, entity_name) 

83 

84 def _get_prov_path(self) -> str: 

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

86 

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

88 entity_name = str(entity_name) 

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

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

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

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

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

94 

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

96 self.__initialize_file_if_not_existing(file_path, entity_name) 

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

98 data = json.load(file) 

99 if entity_name in data: 

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

101 else: 

102 self.prov_files[entity_name] = 0 

103 return self.prov_files[entity_name] 

104 

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

106 self.__initialize_file_if_not_existing(file_path, entity_name) 

107 cur_number = self._read_number(file_path, entity_name) 

108 cur_number += 1 

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

110 data = json.load(file) 

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

112 data[entity_name] = cur_number 

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

114 outfile.write(json_object) 

115 return cur_number