Coverage for oc_ocdm / counter_handler / filesystem_counter_handler.py: 96%
167 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-06 20:05 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com>
4# SPDX-FileCopyrightText: 2024-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
8# -*- coding: utf-8 -*-
9from __future__ import annotations
11import os
12from typing import TYPE_CHECKING
14from filelock import FileLock
16if TYPE_CHECKING:
17 from typing import Dict, List, Tuple
19from oc_ocdm.counter_handler.counter_handler import CounterHandler
20from oc_ocdm.support.support import is_string_empty
23class FilesystemCounterHandler(CounterHandler):
24 """A concrete implementation of the ``CounterHandler`` interface that persistently stores the counter values within the filesystem.
26 Counter data is loaded into RAM on first access per supplier prefix (lazy loading) and written back to disk only when ``flush()`` is called."""
28 def __init__(self, info_dir: str | None, supplier_prefix: str = "") -> None:
29 """
30 Constructor of the ``FilesystemCounterHandler`` class.
32 :param info_dir: The path to the folder that does/will contain the counter values.
33 :type info_dir: str
34 :raises ValueError: if ``info_dir`` is None or an empty string.
35 """
36 if info_dir is None or is_string_empty(info_dir):
37 raise ValueError("info_dir parameter is required!")
39 if info_dir[-1] != os.sep:
40 info_dir += os.sep
42 self.info_dir: str = info_dir
43 self.supplier_prefix: str = supplier_prefix
44 self.datasets_dir: str = info_dir + "datasets" + os.sep
45 self.short_names: List[str] = ["an", "ar", "be", "br", "ci", "de", "id", "pl", "ra", "re", "rp"]
46 self.metadata_short_names: List[str] = ["di"]
47 self.info_files: Dict[str, str] = {key: ("info_file_" + key + ".txt") for key in self.short_names}
48 self.prov_files: Dict[str, str] = {key: ("prov_file_" + key + ".txt") for key in self.short_names}
50 self._cache: Dict[str, List[int]] = {}
51 self._dirty: set[str] = set()
52 self._dirty_lines: Dict[str, set[int]] = {}
53 self._loaded_dirs: set[str] = set()
55 self._ensure_loaded(supplier_prefix)
57 def _get_prefix_dir(self, supplier_prefix: str | None) -> str:
58 sp = "" if supplier_prefix is None else supplier_prefix
59 if sp == self.supplier_prefix or not self.supplier_prefix:
60 return self.info_dir
61 return self.info_dir.replace(self.supplier_prefix, sp, 1)
63 def _ensure_loaded(self, supplier_prefix: str) -> None:
64 prefix_dir = self._get_prefix_dir(supplier_prefix)
65 if prefix_dir in self._loaded_dirs:
66 return
67 if not os.path.isdir(prefix_dir):
68 self._loaded_dirs.add(prefix_dir)
69 return
70 for filename in os.listdir(prefix_dir):
71 if not filename.endswith(".txt"):
72 continue
73 if not (filename.startswith("info_file_") or filename.startswith("prov_file_")):
74 continue
75 filepath = prefix_dir + filename
76 with open(filepath, "r") as f:
77 self._cache[filepath] = [int(line.rstrip("\n")) if line.rstrip("\n") else 0 for line in f]
78 self._loaded_dirs.add(prefix_dir)
80 def flush(self) -> None:
81 for file_path in self._dirty:
82 dir_path = os.path.dirname(file_path)
83 if not os.path.exists(dir_path):
84 os.makedirs(dir_path)
85 cache_list = self._cache[file_path]
86 dirty_lines = self._dirty_lines[file_path]
87 # Read-modify-write under a lock so concurrent handlers sharing this info_dir
88 # do not overwrite one another's counters: only the lines this handler changed
89 # are written, every other line is preserved from the current on-disk content.
90 with FileLock(file_path + ".lock"):
91 merged = self._read_persisted_counters(file_path)
92 if len(merged) < len(cache_list):
93 merged.extend([0] * (len(cache_list) - len(merged)))
94 for line_index in dirty_lines:
95 merged[line_index] = cache_list[line_index]
96 with open(file_path, "w") as f:
97 f.writelines(f"{v}\n" if v else "\n" for v in merged)
98 self._cache[file_path] = merged
99 self._dirty.clear()
100 self._dirty_lines.clear()
102 @staticmethod
103 def _read_persisted_counters(file_path: str) -> List[int]:
104 if not os.path.exists(file_path):
105 return []
106 with open(file_path, "r") as f:
107 return [int(line.rstrip("\n")) if line.rstrip("\n") else 0 for line in f]
109 def set_counter(
110 self,
111 new_value: int,
112 entity_short_name: str,
113 prov_short_name: str = "",
114 identifier: int = 1,
115 supplier_prefix: str = "",
116 ) -> None:
117 """
118 It allows to set the counter value of graph and provenance entities.
120 :param new_value: The new counter value to be set
121 :type new_value: int
122 :param entity_short_name: The short name associated either to the type of the entity itself
123 or, in case of a provenance entity, to the type of the relative graph entity.
124 :type entity_short_name: str
125 :param prov_short_name: In case of a provenance entity, the short name associated to the type
126 of the entity itself. An empty string otherwise.
127 :type prov_short_name: str
128 :param identifier: In case of a provenance entity, the counter value that identifies the relative
129 graph entity. The integer value '1' otherwise.
130 :type identifier: int
131 :raises ValueError: if ``new_value`` is a negative integer or ``identifier`` is less than or equal to zero.
132 :return: None
133 """
134 if new_value < 0:
135 raise ValueError("new_value must be a non negative integer!")
136 self._ensure_loaded(supplier_prefix)
137 if prov_short_name == "se":
138 file_path: str = self._get_prov_path(entity_short_name, supplier_prefix)
139 else:
140 file_path: str = self._get_info_path(entity_short_name, supplier_prefix)
141 self._set_number(new_value, file_path, identifier)
143 def set_counters_batch(self, updates: Dict[Tuple[str, str], Dict[int, int]], supplier_prefix: str) -> None:
144 """
145 Updates counters in batch for multiple files.
146 `updates` is a dictionary where the key is a tuple (entity_short_name, prov_short_name)
147 and the value is a dictionary of line numbers to new counter values.
148 """
149 self._ensure_loaded(supplier_prefix)
150 for (entity_short_name, prov_short_name), file_updates in updates.items():
151 file_path = (
152 self._get_prov_path(entity_short_name, supplier_prefix)
153 if prov_short_name == "se"
154 else self._get_info_path(entity_short_name, supplier_prefix)
155 )
156 self._set_numbers(file_path, file_updates)
158 def _set_numbers(self, file_path: str, updates: Dict[int, int]) -> None:
159 """
160 Apply multiple counter updates to a single file.
161 `updates` is a dictionary where the key is the line number (identifier)
162 and the value is the new counter value.
163 """
164 if file_path not in self._cache:
165 self._cache[file_path] = [0]
166 cache_list = self._cache[file_path]
167 needed = max(updates.keys()) + 1 - len(cache_list)
168 if needed > 0:
169 cache_list.extend([0] * needed)
170 dirty_lines = self._dirty_lines.setdefault(file_path, set())
171 for line_number, new_value in updates.items():
172 cache_list[line_number - 1] = new_value
173 dirty_lines.add(line_number - 1)
174 self._dirty.add(file_path)
176 def read_counter(
177 self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = ""
178 ) -> int:
179 """
180 It allows to read the counter value of graph and provenance entities.
182 :param entity_short_name: The short name associated either to the type of the entity itself
183 or, in case of a provenance entity, to the type of the relative graph entity.
184 :type entity_short_name: str
185 :param prov_short_name: In case of a provenance entity, the short name associated to the type
186 of the entity itself. An empty string otherwise.
187 :type prov_short_name: str
188 :param identifier: In case of a provenance entity, the counter value that identifies the relative
189 graph entity. The integer value '1' otherwise.
190 :type identifier: int
191 :raises ValueError: if ``identifier`` is less than or equal to zero.
192 :return: The requested counter value.
193 """
194 self._ensure_loaded(supplier_prefix)
195 if prov_short_name == "se":
196 file_path: str = self._get_prov_path(entity_short_name, supplier_prefix)
197 else:
198 file_path: str = self._get_info_path(entity_short_name, supplier_prefix)
199 return self._read_number(file_path, identifier)
201 def increment_counter(
202 self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = ""
203 ) -> int:
204 """
205 It allows to increment the counter value of graph and provenance entities by one unit.
207 :param entity_short_name: The short name associated either to the type of the entity itself
208 or, in case of a provenance entity, to the type of the relative graph entity.
209 :type entity_short_name: str
210 :param prov_short_name: In case of a provenance entity, the short name associated to the type
211 of the entity itself. An empty string otherwise.
212 :type prov_short_name: str
213 :param identifier: In case of a provenance entity, the counter value that identifies the relative
214 graph entity. The integer value '1' otherwise.
215 :type identifier: int
216 :raises ValueError: if ``identifier`` is less than or equal to zero.
217 :return: The newly-updated (already incremented) counter value.
218 """
219 self._ensure_loaded(supplier_prefix)
220 if prov_short_name == "se":
221 file_path: str = self._get_prov_path(entity_short_name, supplier_prefix)
222 else:
223 file_path: str = self._get_info_path(entity_short_name, supplier_prefix)
224 return self._add_number(file_path, identifier)
226 def _get_info_path(self, short_name: str, supplier_prefix: str) -> str:
227 return self._get_prefix_dir(supplier_prefix) + self.info_files[short_name]
229 def _get_prov_path(self, short_name: str, supplier_prefix: str) -> str:
230 return self._get_prefix_dir(supplier_prefix) + self.prov_files[short_name]
232 def _get_metadata_path(self, short_name: str, dataset_name: str) -> str:
233 return self.datasets_dir + dataset_name + os.sep + "metadata_" + short_name + ".txt"
235 def _read_number(self, file_path: str, line_number: int) -> int:
236 if line_number <= 0:
237 raise ValueError("line_number must be a positive non-zero integer number!")
238 if file_path in self._cache:
239 idx = line_number - 1
240 cache_list = self._cache[file_path]
241 if idx < len(cache_list):
242 return cache_list[idx]
243 return 0
244 self._cache[file_path] = [0]
245 return 0
247 def _add_number(self, file_path: str, line_number: int = 1) -> int:
248 if line_number <= 0:
249 raise ValueError("line_number must be a positive non-zero integer number!")
250 current_value = self._read_number(file_path, line_number)
251 new_value = current_value + 1
252 self._set_number(new_value, file_path, line_number)
253 return new_value
255 def _set_number(self, new_value: int, file_path: str, line_number: int = 1) -> None:
256 if new_value < 0:
257 raise ValueError("new_value must be a non negative integer!")
258 if line_number <= 0:
259 raise ValueError("line_number must be a positive non-zero integer number!")
260 if file_path not in self._cache:
261 self._cache[file_path] = [0]
262 cache_list = self._cache[file_path]
263 needed = line_number - len(cache_list)
264 if needed > 0:
265 cache_list.extend([0] * needed)
266 cache_list[line_number - 1] = new_value
267 self._dirty.add(file_path)
268 self._dirty_lines.setdefault(file_path, set()).add(line_number - 1)
270 def set_metadata_counter(self, new_value: int, entity_short_name: str, dataset_name: str | None) -> None:
271 """
272 It allows to set the counter value of metadata entities.
274 :param new_value: The new counter value to be set
275 :type new_value: int
276 :param entity_short_name: The short name associated either to the type of the entity itself.
277 :type entity_short_name: str
278 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
279 :type dataset_name: str
280 :raises ValueError: if ``new_value`` is a negative integer, ``dataset_name`` is None or
281 ``entity_short_name`` is not a known metadata short name.
282 :return: None
283 """
284 if new_value < 0:
285 raise ValueError("new_value must be a non negative integer!")
286 if dataset_name is None:
287 raise ValueError("dataset_name must be provided!")
288 if entity_short_name not in self.metadata_short_names:
289 raise ValueError("entity_short_name is not a known metadata short name!")
290 file_path: str = self._get_metadata_path(entity_short_name, dataset_name)
291 return self._set_number(new_value, file_path, 1)
293 def read_metadata_counter(self, entity_short_name: str, dataset_name: str | None) -> int:
294 """
295 It allows to read the counter value of metadata entities.
297 :param entity_short_name: The short name associated either to the type of the entity itself.
298 :type entity_short_name: str
299 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
300 :type dataset_name: str
301 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
302 :return: The requested counter value.
303 """
304 if dataset_name is None:
305 raise ValueError("dataset_name must be provided!")
306 if entity_short_name not in self.metadata_short_names:
307 raise ValueError("entity_short_name is not a known metadata short name!")
308 file_path: str = self._get_metadata_path(entity_short_name, dataset_name)
309 return self._read_number(file_path, 1)
311 def increment_metadata_counter(self, entity_short_name: str, dataset_name: str | None) -> int:
312 """
313 It allows to increment the counter value of metadata entities by one unit.
315 :param entity_short_name: The short name associated either to the type of the entity itself.
316 :type entity_short_name: str
317 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
318 :type dataset_name: str
319 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
320 :return: The newly-updated (already incremented) counter value.
321 """
322 if dataset_name is None:
323 raise ValueError("dataset_name must be provided!")
324 if entity_short_name not in self.metadata_short_names:
325 raise ValueError("entity_short_name is not a known metadata short name!")
326 file_path: str = self._get_metadata_path(entity_short_name, dataset_name)
327 return self._add_number(file_path, 1)