Coverage for oc_ocdm / counter_handler / in_memory_counter_handler.py: 94%
89 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 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
8# -*- coding: utf-8 -*-
9from __future__ import annotations
11from typing import TYPE_CHECKING
13if TYPE_CHECKING:
14 from typing import Dict, List
16from oc_ocdm.counter_handler.counter_handler import CounterHandler
19class InMemoryCounterHandler(CounterHandler):
20 """A concrete implementation of the ``CounterHandler`` interface that temporarily stores
21 the counter values in the volatile system memory."""
23 def __init__(self) -> None:
24 """
25 Constructor of the ``InMemoryCounterHandler`` class.
26 """
27 self.short_names: List[str] = ["an", "ar", "be", "br", "ci", "de", "id", "pl", "ra", "re", "rp"]
28 self.prov_short_names: List[str] = ["se"]
29 self.metadata_short_names: List[str] = ["di"]
30 self.entity_counters: Dict[str, int] = {key: 0 for key in self.short_names}
31 self.prov_counters: Dict[str, Dict[str, List[int]]] = {
32 key1: {key2: [] for key2 in self.prov_short_names} for key1 in self.short_names
33 }
34 self.metadata_counters: Dict[str, Dict[str, int]] = {}
36 def set_counter(
37 self,
38 new_value: int,
39 entity_short_name: str,
40 prov_short_name: str = "",
41 identifier: int = 1,
42 supplier_prefix: str = "",
43 ) -> None:
44 """
45 It allows to set the counter value of graph and provenance entities.
47 :param new_value: The new counter value to be set
48 :type new_value: int
49 :param entity_short_name: The short name associated either to the type of the entity itself
50 or, in case of a provenance entity, to the type of the relative graph entity.
51 :type entity_short_name: str
52 :param prov_short_name: In case of a provenance entity, the short name associated to the type
53 of the entity itself. An empty string otherwise.
54 :type prov_short_name: str
55 :param identifier: In case of a provenance entity, the counter value that identifies the relative
56 graph entity. The integer value '1' otherwise.
57 :type identifier: int
58 :raises ValueError: if ``new_value`` is a negative integer, ``identifier`` is less than or equal to zero,
59 ``entity_short_name`` is not a known short name or ``prov_short_name`` is not a known provenance short name.
60 :return: None
61 """
62 if new_value < 0:
63 raise ValueError("new_value must be a non negative integer!")
65 if entity_short_name not in self.short_names:
66 raise ValueError("entity_short_name is not a known short name!")
68 if prov_short_name != "":
69 if prov_short_name not in self.prov_short_names:
70 raise ValueError("prov_short_name is not a known provenance short name!")
71 if identifier <= 0:
72 raise ValueError("identifier must be a positive non-zero integer number!")
74 identifier -= 1 # Internally we use zero_indexing!
75 if prov_short_name in self.prov_short_names:
76 # It's a provenance entity!
77 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
78 if missing_counters > 0:
79 self.prov_counters[entity_short_name][prov_short_name] += [0] * missing_counters
80 self.prov_counters[entity_short_name][prov_short_name][identifier] = new_value
81 else:
82 # It's an entity!
83 self.entity_counters[entity_short_name] = new_value
85 def read_counter(
86 self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = ""
87 ) -> int:
88 """
89 It allows to read the counter value of graph and provenance entities.
91 :param entity_short_name: The short name associated either to the type of the entity itself
92 or, in case of a provenance entity, to the type of the relative graph entity.
93 :type entity_short_name: str
94 :param prov_short_name: In case of a provenance entity, the short name associated to the type
95 of the entity itself. An empty string otherwise.
96 :type prov_short_name: str
97 :param identifier: In case of a provenance entity, the counter value that identifies the relative
98 graph entity. The integer value '1' otherwise.
99 :type identifier: int
100 :raises ValueError: if ``identifier`` is less than or equal to zero, ``entity_short_name``
101 is not a known short name or ``prov_short_name`` is not a known provenance short name.
102 :return: The requested counter value.
103 """
104 if entity_short_name not in self.short_names:
105 raise ValueError("entity_short_name is not a known short name!")
107 if prov_short_name != "":
108 if prov_short_name not in self.prov_short_names:
109 raise ValueError("prov_short_name is not a known provenance short name!")
110 if identifier <= 0:
111 raise ValueError("identifier must be a positive non-zero integer number!")
113 identifier -= 1 # Internally we use zero_indexing!
114 if prov_short_name in self.prov_short_names:
115 # It's a provenance entity!
116 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
117 if missing_counters > 0:
118 self.prov_counters[entity_short_name][prov_short_name] += [0] * missing_counters
119 return self.prov_counters[entity_short_name][prov_short_name][identifier]
120 else:
121 # It's an entity!
122 return self.entity_counters[entity_short_name]
124 def increment_counter(
125 self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = ""
126 ) -> int:
127 """
128 It allows to increment the counter value of graph and provenance entities by one unit.
130 :param entity_short_name: The short name associated either to the type of the entity itself
131 or, in case of a provenance entity, to the type of the relative graph entity.
132 :type entity_short_name: str
133 :param prov_short_name: In case of a provenance entity, the short name associated to the type
134 of the entity itself. An empty string otherwise.
135 :type prov_short_name: str
136 :param identifier: In case of a provenance entity, the counter value that identifies the relative
137 graph entity. The integer value '1' otherwise.
138 :type identifier: int
139 :raises ValueError: if ``identifier`` is less than or equal to zero, ``entity_short_name``
140 is not a known short name or ``prov_short_name`` is not a known provenance short name.
141 :return: The newly-updated (already incremented) counter value.
142 """
143 if entity_short_name not in self.short_names:
144 raise ValueError("entity_short_name is not a known short name!")
146 if prov_short_name != "":
147 if prov_short_name not in self.prov_short_names:
148 raise ValueError("prov_short_name is not a known provenance short name!")
149 if identifier <= 0:
150 raise ValueError("identifier must be a positive non-zero integer number!")
152 identifier -= 1 # Internally we use zero_indexing!
153 if prov_short_name in self.prov_short_names:
154 # It's a provenance entity!
155 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
156 if missing_counters > 0:
157 self.prov_counters[entity_short_name][prov_short_name] += [0] * missing_counters
158 self.prov_counters[entity_short_name][prov_short_name][identifier] += 1
159 return self.prov_counters[entity_short_name][prov_short_name][identifier]
160 else:
161 # It's an entity!
162 self.entity_counters[entity_short_name] += 1
163 return self.entity_counters[entity_short_name]
165 def set_metadata_counter(self, new_value: int, entity_short_name: str, dataset_name: str | None) -> None:
166 """
167 It allows to set the counter value of metadata entities.
169 :param new_value: The new counter value to be set
170 :type new_value: int
171 :param entity_short_name: The short name associated either to the type of the entity itself.
172 :type entity_short_name: str
173 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
174 :type dataset_name: str
175 :raises ValueError: if ``new_value`` is a negative integer, ``dataset_name`` is None or
176 ``entity_short_name`` is not a known metadata short name.
177 :return: None
178 """
179 if new_value < 0:
180 raise ValueError("new_value must be a non negative integer!")
182 if dataset_name is None:
183 raise ValueError("dataset_name must be provided!")
185 if entity_short_name not in self.metadata_short_names:
186 raise ValueError("entity_short_name is not a known metadata short name!")
188 if dataset_name not in self.metadata_counters:
189 self.metadata_counters[dataset_name] = {key: 0 for key in self.metadata_short_names}
191 self.metadata_counters[dataset_name][entity_short_name] = new_value
193 def read_metadata_counter(self, entity_short_name: str, dataset_name: str | None) -> int:
194 """
195 It allows to read the counter value of metadata entities.
197 :param entity_short_name: The short name associated either to the type of the entity itself.
198 :type entity_short_name: str
199 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
200 :type dataset_name: str
201 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
202 :return: The requested counter value.
203 """
204 if dataset_name is None:
205 raise ValueError("dataset_name must be provided!")
207 if entity_short_name not in self.metadata_short_names:
208 raise ValueError("entity_short_name is not a known metadata short name!")
210 if dataset_name not in self.metadata_counters:
211 return 0
212 else:
213 if entity_short_name not in self.metadata_counters[dataset_name]:
214 return 0
215 else:
216 return self.metadata_counters[dataset_name][entity_short_name]
218 def increment_metadata_counter(self, entity_short_name: str, dataset_name: str | None) -> int:
219 """
220 It allows to increment the counter value of metadata entities by one unit.
222 :param entity_short_name: The short name associated either to the type of the entity itself.
223 :type entity_short_name: str
224 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
225 :type dataset_name: str
226 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
227 :return: The newly-updated (already incremented) counter value.
228 """
229 if dataset_name is None:
230 raise ValueError("dataset_name must be provided!")
232 if entity_short_name not in self.metadata_short_names:
233 raise ValueError("entity_short_name is not a known metadata short name!")
235 if dataset_name not in self.metadata_counters:
236 self.metadata_counters[dataset_name] = {key: 0 for key in self.metadata_short_names}
238 self.metadata_counters[dataset_name][entity_short_name] += 1
239 return self.metadata_counters[dataset_name][entity_short_name]