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-03-28 18:52 +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 List, Dict
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]]] = {key1: {key2: [] for key2 in self.prov_short_names}
32 for key1 in self.short_names}
33 self.metadata_counters: Dict[str, Dict[str, int]] = {}
35 def set_counter(self, new_value: int, entity_short_name: str, prov_short_name: str = "",
36 identifier: int = 1, supplier_prefix: str = "") -> None:
37 """
38 It allows to set the counter value of graph and provenance entities.
40 :param new_value: The new counter value to be set
41 :type new_value: int
42 :param entity_short_name: The short name associated either to the type of the entity itself
43 or, in case of a provenance entity, to the type of the relative graph entity.
44 :type entity_short_name: str
45 :param prov_short_name: In case of a provenance entity, the short name associated to the type
46 of the entity itself. An empty string otherwise.
47 :type prov_short_name: str
48 :param identifier: In case of a provenance entity, the counter value that identifies the relative
49 graph entity. The integer value '1' otherwise.
50 :type identifier: int
51 :raises ValueError: if ``new_value`` is a negative integer, ``identifier`` is less than or equal to zero,
52 ``entity_short_name`` is not a known short name or ``prov_short_name`` is not a known provenance short name.
53 :return: None
54 """
55 if new_value < 0:
56 raise ValueError("new_value must be a non negative integer!")
58 if entity_short_name not in self.short_names:
59 raise ValueError("entity_short_name is not a known short name!")
61 if prov_short_name != "":
62 if prov_short_name not in self.prov_short_names:
63 raise ValueError("prov_short_name is not a known provenance short name!")
64 if identifier <= 0:
65 raise ValueError("identifier must be a positive non-zero integer number!")
67 identifier -= 1 # Internally we use zero_indexing!
68 if prov_short_name in self.prov_short_names:
69 # It's a provenance entity!
70 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
71 if missing_counters > 0:
72 self.prov_counters[entity_short_name][prov_short_name] += [0] * missing_counters
73 self.prov_counters[entity_short_name][prov_short_name][identifier] = new_value
74 else:
75 # It's an entity!
76 self.entity_counters[entity_short_name] = new_value
78 def read_counter(self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = "") -> int:
79 """
80 It allows to read the counter value of graph and provenance entities.
82 :param entity_short_name: The short name associated either to the type of the entity itself
83 or, in case of a provenance entity, to the type of the relative graph entity.
84 :type entity_short_name: str
85 :param prov_short_name: In case of a provenance entity, the short name associated to the type
86 of the entity itself. An empty string otherwise.
87 :type prov_short_name: str
88 :param identifier: In case of a provenance entity, the counter value that identifies the relative
89 graph entity. The integer value '1' otherwise.
90 :type identifier: int
91 :raises ValueError: if ``identifier`` is less than or equal to zero, ``entity_short_name``
92 is not a known short name or ``prov_short_name`` is not a known provenance short name.
93 :return: The requested counter value.
94 """
95 if entity_short_name not in self.short_names:
96 raise ValueError("entity_short_name is not a known short name!")
98 if prov_short_name != "":
99 if prov_short_name not in self.prov_short_names:
100 raise ValueError("prov_short_name is not a known provenance short name!")
101 if identifier <= 0:
102 raise ValueError("identifier must be a positive non-zero integer number!")
104 identifier -= 1 # Internally we use zero_indexing!
105 if prov_short_name in self.prov_short_names:
106 # It's a provenance entity!
107 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
108 if missing_counters > 0:
109 self.prov_counters[entity_short_name][prov_short_name] += [0] * missing_counters
110 return self.prov_counters[entity_short_name][prov_short_name][identifier]
111 else:
112 # It's an entity!
113 return self.entity_counters[entity_short_name]
115 def increment_counter(self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = "") -> int:
116 """
117 It allows to increment the counter value of graph and provenance entities by one unit.
119 :param entity_short_name: The short name associated either to the type of the entity itself
120 or, in case of a provenance entity, to the type of the relative graph entity.
121 :type entity_short_name: str
122 :param prov_short_name: In case of a provenance entity, the short name associated to the type
123 of the entity itself. An empty string otherwise.
124 :type prov_short_name: str
125 :param identifier: In case of a provenance entity, the counter value that identifies the relative
126 graph entity. The integer value '1' otherwise.
127 :type identifier: int
128 :raises ValueError: if ``identifier`` is less than or equal to zero, ``entity_short_name``
129 is not a known short name or ``prov_short_name`` is not a known provenance short name.
130 :return: The newly-updated (already incremented) counter value.
131 """
132 if entity_short_name not in self.short_names:
133 raise ValueError("entity_short_name is not a known short name!")
135 if prov_short_name != "":
136 if prov_short_name not in self.prov_short_names:
137 raise ValueError("prov_short_name is not a known provenance short name!")
138 if identifier <= 0:
139 raise ValueError("identifier must be a positive non-zero integer number!")
141 identifier -= 1 # Internally we use zero_indexing!
142 if prov_short_name in self.prov_short_names:
143 # It's a provenance entity!
144 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
145 if missing_counters > 0:
146 self.prov_counters[entity_short_name][prov_short_name] += [0]*missing_counters
147 self.prov_counters[entity_short_name][prov_short_name][identifier] += 1
148 return self.prov_counters[entity_short_name][prov_short_name][identifier]
149 else:
150 # It's an entity!
151 self.entity_counters[entity_short_name] += 1
152 return self.entity_counters[entity_short_name]
154 def set_metadata_counter(self, new_value: int, entity_short_name: str, dataset_name: str) -> None:
155 """
156 It allows to set the counter value of metadata entities.
158 :param new_value: The new counter value to be set
159 :type new_value: int
160 :param entity_short_name: The short name associated either to the type of the entity itself.
161 :type entity_short_name: str
162 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
163 :type dataset_name: str
164 :raises ValueError: if ``new_value`` is a negative integer, ``dataset_name`` is None or
165 ``entity_short_name`` is not a known metadata short name.
166 :return: None
167 """
168 if new_value < 0:
169 raise ValueError("new_value must be a non negative integer!")
171 if dataset_name is None:
172 raise ValueError("dataset_name must be provided!")
174 if entity_short_name not in self.metadata_short_names:
175 raise ValueError("entity_short_name is not a known metadata short name!")
177 if dataset_name not in self.metadata_counters:
178 self.metadata_counters[dataset_name] = {key: 0 for key in self.metadata_short_names}
180 self.metadata_counters[dataset_name][entity_short_name] = new_value
182 def read_metadata_counter(self, entity_short_name: str, dataset_name: str) -> int:
183 """
184 It allows to read the counter value of metadata entities.
186 :param entity_short_name: The short name associated either to the type of the entity itself.
187 :type entity_short_name: str
188 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
189 :type dataset_name: str
190 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
191 :return: The requested counter value.
192 """
193 if dataset_name is None:
194 raise ValueError("dataset_name must be provided!")
196 if entity_short_name not in self.metadata_short_names:
197 raise ValueError("entity_short_name is not a known metadata short name!")
199 if dataset_name not in self.metadata_counters:
200 return 0
201 else:
202 if entity_short_name not in self.metadata_counters[dataset_name]:
203 return 0
204 else:
205 return self.metadata_counters[dataset_name][entity_short_name]
207 def increment_metadata_counter(self, entity_short_name: str, dataset_name: str) -> int:
208 """
209 It allows to increment the counter value of metadata entities by one unit.
211 :param entity_short_name: The short name associated either to the type of the entity itself.
212 :type entity_short_name: str
213 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
214 :type dataset_name: str
215 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
216 :return: The newly-updated (already incremented) counter value.
217 """
218 if dataset_name is None:
219 raise ValueError("dataset_name must be provided!")
221 if entity_short_name not in self.metadata_short_names:
222 raise ValueError("entity_short_name is not a known metadata short name!")
224 if dataset_name not in self.metadata_counters:
225 self.metadata_counters[dataset_name] = {key: 0 for key in self.metadata_short_names}
227 self.metadata_counters[dataset_name][entity_short_name] += 1
228 return self.metadata_counters[dataset_name][entity_short_name]