Coverage for oc_ocdm/counter_handler/in_memory_counter_handler.py: 92%
91 statements
« prev ^ index » next coverage.py v6.5.0, created at 2025-05-30 22:05 +0000
1#!/usr/bin/python
2# -*- coding: utf-8 -*-
3# Copyright (c) 2016, Silvio Peroni <essepuntato@gmail.com>
4#
5# Permission to use, copy, modify, and/or distribute this software for any purpose
6# with or without fee is hereby granted, provided that the above copyright notice
7# and this permission notice appear in all copies.
8#
9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT,
12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE,
13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
15# SOFTWARE.
16from __future__ import annotations
18from typing import TYPE_CHECKING
20if TYPE_CHECKING:
21 from typing import List, Dict
23from oc_ocdm.counter_handler.counter_handler import CounterHandler
26class InMemoryCounterHandler(CounterHandler):
27 """A concrete implementation of the ``CounterHandler`` interface that temporarily stores
28 the counter values in the volatile system memory."""
30 def __init__(self) -> None:
31 """
32 Constructor of the ``InMemoryCounterHandler`` class.
33 """
34 self.short_names: List[str] = ["an", "ar", "be", "br", "ci", "de", "id", "pl", "ra", "re", "rp"]
35 self.prov_short_names: List[str] = ["se"]
36 self.metadata_short_names: List[str] = ["di"]
37 self.entity_counters: Dict[str, int] = {key: 0 for key in self.short_names}
38 self.prov_counters: Dict[str, Dict[str, List[int]]] = {key1: {key2: [] for key2 in self.prov_short_names}
39 for key1 in self.short_names}
40 self.metadata_counters: Dict[str, Dict[str, int]] = {}
42 def set_counter(self, new_value: int, entity_short_name: str, prov_short_name: str = "",
43 identifier: int = 1, supplier_prefix: str = "") -> 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(self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = "") -> int:
86 """
87 It allows to read the counter value of graph and provenance entities.
89 :param entity_short_name: The short name associated either to the type of the entity itself
90 or, in case of a provenance entity, to the type of the relative graph entity.
91 :type entity_short_name: str
92 :param prov_short_name: In case of a provenance entity, the short name associated to the type
93 of the entity itself. An empty string otherwise.
94 :type prov_short_name: str
95 :param identifier: In case of a provenance entity, the counter value that identifies the relative
96 graph entity. The integer value '1' otherwise.
97 :type identifier: int
98 :raises ValueError: if ``identifier`` is less than or equal to zero, ``entity_short_name``
99 is not a known short name or ``prov_short_name`` is not a known provenance short name.
100 :return: The requested counter value.
101 """
102 if entity_short_name not in self.short_names:
103 raise ValueError("entity_short_name is not a known short name!")
105 if prov_short_name != "":
106 if prov_short_name not in self.prov_short_names:
107 raise ValueError("prov_short_name is not a known provenance short name!")
108 if identifier <= 0:
109 raise ValueError("identifier must be a positive non-zero integer number!")
111 identifier -= 1 # Internally we use zero_indexing!
112 if prov_short_name in self.prov_short_names:
113 # It's a provenance entity!
114 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
115 if missing_counters > 0:
116 self.prov_counters[entity_short_name][prov_short_name] += [0] * missing_counters
117 return self.prov_counters[entity_short_name][prov_short_name][identifier]
118 else:
119 # It's an entity!
120 return self.entity_counters[entity_short_name]
122 def increment_counter(self, entity_short_name: str, prov_short_name: str = "", identifier: int = 1, supplier_prefix: str = "") -> int:
123 """
124 It allows to increment the counter value of graph and provenance entities by one unit.
126 :param entity_short_name: The short name associated either to the type of the entity itself
127 or, in case of a provenance entity, to the type of the relative graph entity.
128 :type entity_short_name: str
129 :param prov_short_name: In case of a provenance entity, the short name associated to the type
130 of the entity itself. An empty string otherwise.
131 :type prov_short_name: str
132 :param identifier: In case of a provenance entity, the counter value that identifies the relative
133 graph entity. The integer value '1' otherwise.
134 :type identifier: int
135 :raises ValueError: if ``identifier`` is less than or equal to zero, ``entity_short_name``
136 is not a known short name or ``prov_short_name`` is not a known provenance short name.
137 :return: The newly-updated (already incremented) counter value.
138 """
139 if entity_short_name not in self.short_names:
140 raise ValueError("entity_short_name is not a known short name!")
142 if prov_short_name != "":
143 if prov_short_name not in self.prov_short_names:
144 raise ValueError("prov_short_name is not a known provenance short name!")
145 if identifier <= 0:
146 raise ValueError("identifier must be a positive non-zero integer number!")
148 identifier -= 1 # Internally we use zero_indexing!
149 if prov_short_name in self.prov_short_names:
150 # It's a provenance entity!
151 missing_counters: int = identifier - (len(self.prov_counters[entity_short_name][prov_short_name]) - 1)
152 if missing_counters > 0:
153 self.prov_counters[entity_short_name][prov_short_name] += [0]*missing_counters
154 self.prov_counters[entity_short_name][prov_short_name][identifier] += 1
155 return self.prov_counters[entity_short_name][prov_short_name][identifier]
156 else:
157 # It's an entity!
158 self.entity_counters[entity_short_name] += 1
159 return self.entity_counters[entity_short_name]
161 def set_metadata_counter(self, new_value: int, entity_short_name: str, dataset_name: str) -> None:
162 """
163 It allows to set the counter value of metadata entities.
165 :param new_value: The new counter value to be set
166 :type new_value: int
167 :param entity_short_name: The short name associated either to the type of the entity itself.
168 :type entity_short_name: str
169 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
170 :type dataset_name: str
171 :raises ValueError: if ``new_value`` is a negative integer, ``dataset_name`` is None or
172 ``entity_short_name`` is not a known metadata short name.
173 :return: None
174 """
175 if new_value < 0:
176 raise ValueError("new_value must be a non negative integer!")
178 if dataset_name is None:
179 raise ValueError("dataset_name must be provided!")
181 if entity_short_name not in self.metadata_short_names:
182 raise ValueError("entity_short_name is not a known metadata short name!")
184 if dataset_name not in self.metadata_counters:
185 self.metadata_counters[dataset_name] = {key: 0 for key in self.metadata_short_names}
187 self.metadata_counters[dataset_name][entity_short_name] = new_value
189 def read_metadata_counter(self, entity_short_name: str, dataset_name: str) -> int:
190 """
191 It allows to read the counter value of metadata entities.
193 :param entity_short_name: The short name associated either to the type of the entity itself.
194 :type entity_short_name: str
195 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
196 :type dataset_name: str
197 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
198 :return: The requested counter value.
199 """
200 if dataset_name is None:
201 raise ValueError("dataset_name must be provided!")
203 if entity_short_name not in self.metadata_short_names:
204 raise ValueError("entity_short_name is not a known metadata short name!")
206 if dataset_name not in self.metadata_counters:
207 return 0
208 else:
209 if entity_short_name not in self.metadata_counters[dataset_name]:
210 return 0
211 else:
212 return self.metadata_counters[dataset_name][entity_short_name]
214 def increment_metadata_counter(self, entity_short_name: str, dataset_name: str) -> int:
215 """
216 It allows to increment the counter value of metadata entities by one unit.
218 :param entity_short_name: The short name associated either to the type of the entity itself.
219 :type entity_short_name: str
220 :param dataset_name: In case of a ``Dataset``, its name. Otherwise, the name of the relative dataset.
221 :type dataset_name: str
222 :raises ValueError: if ``dataset_name`` is None or ``entity_short_name`` is not a known metadata short name.
223 :return: The newly-updated (already incremented) counter value.
224 """
225 if dataset_name is None:
226 raise ValueError("dataset_name must be provided!")
228 if entity_short_name not in self.metadata_short_names:
229 raise ValueError("entity_short_name is not a known metadata short name!")
231 if dataset_name not in self.metadata_counters:
232 self.metadata_counters[dataset_name] = {key: 0 for key in self.metadata_short_names}
234 self.metadata_counters[dataset_name][entity_short_name] += 1
235 return self.metadata_counters[dataset_name][entity_short_name]