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-2026 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 

18 that persistently stores the counter values within the 

19 filesystem.""" 

20 

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

22 """ 

23 Constructor of the ``FilesystemCounterHandler`` class. 

24 

25 :param info_dir: The path to the folder that does/will 

26 contain the counter values. 

27 :type info_dir: str 

28 :raises ValueError: if ``info_dir`` is None or an empty 

29 string. 

30 """ 

31 if info_dir is None or is_string_empty(info_dir): 

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

33 

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

35 info_dir += os.sep 

36 

37 self.info_dir: str = info_dir 

38 self.prov_files = dict() 

39 self.provenance_index_filename = "provenance_index.json" 

40 

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

42 """ 

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

44 

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

46 :type new_value: int 

47 :param entity_name: The entity name 

48 :type entity_name: str 

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

50 :return: None 

51 """ 

52 entity_name = str(entity_name) 

53 if new_value < 0: 

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

55 file_path: str = self._get_prov_path() 

56 self.__initialize_file_if_not_existing(file_path, entity_name) 

57 with open(file_path, "r", encoding="utf8") as file: 

58 data = json.load(file) 

59 with open(file_path, "w", encoding="utf8") as outfile: 

60 data[entity_name] = new_value 

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

62 

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

64 """ 

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

66 

67 :param entity_name: The entity name 

68 :type entity_name: str 

69 :return: The requested counter value. 

70 """ 

71 entity_name = str(entity_name) 

72 file_path: str = self._get_prov_path() 

73 return self._read_number(file_path, entity_name) 

74 

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

76 """ 

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

78 

79 :param entity_name: The entity name 

80 :type entity_name: str 

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

82 """ 

83 entity_name = str(entity_name) 

84 file_path: str = self._get_prov_path() 

85 return self._add_number(file_path, entity_name) 

86 

87 def _get_prov_path(self) -> str: 

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

89 

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

91 entity_name = str(entity_name) 

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

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

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

95 with open(file_path, "w", encoding="utf8") as outfile: 

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

97 

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

99 self.__initialize_file_if_not_existing(file_path, entity_name) 

100 with open(file_path, "r", encoding="utf8") as file: 

101 data = json.load(file) 

102 if entity_name in data: 

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

104 else: 

105 self.prov_files[entity_name] = 0 

106 return self.prov_files[entity_name] 

107 

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

109 self.__initialize_file_if_not_existing(file_path, entity_name) 

110 cur_number = self._read_number(file_path, entity_name) 

111 cur_number += 1 

112 with open(file_path, "r", encoding="utf8") as file: 

113 data = json.load(file) 

114 with open(file_path, "w", encoding="utf8") as outfile: 

115 data[entity_name] = cur_number 

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

117 outfile.write(json_object) 

118 return cur_number