Coverage for heritrace/meta_counter_handler.py: 100%
35 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-04-18 11:10 +0000
1import urllib.parse
2import redis
5class MetaCounterHandler:
6 def __init__(self, host='localhost', port=6379, db=0, password=None) -> None:
7 """
8 Constructor of the ``MetaCounterHandler`` class.
10 :param host: Redis host address
11 :type host: str
12 :param port: Redis port number
13 :type port: int
14 :param db: Redis database number
15 :type db: int
16 :param password: Redis password if required
17 :type password: str
18 """
19 self.redis_client = redis.Redis(host=host, port=port, db=db, password=password)
21 self.base_iri = "https://w3id.org/oc/meta/"
22 self.short_names = ["ar", "br", "id", "ra", "re"]
23 self.supplier_prefix = "060"
25 self.entity_type_abbr = {
26 "http://purl.org/spar/fabio/Expression": "br",
27 "http://purl.org/spar/fabio/Article": "br",
28 "http://purl.org/spar/fabio/JournalArticle": "br",
29 "http://purl.org/spar/fabio/Book": "br",
30 "http://purl.org/spar/fabio/BookChapter": "br",
31 "http://purl.org/spar/fabio/JournalIssue": "br",
32 "http://purl.org/spar/fabio/JournalVolume": "br",
33 "http://purl.org/spar/fabio/Journal": "br",
34 "http://purl.org/spar/fabio/AcademicProceedings": "br",
35 "http://purl.org/spar/fabio/ProceedingsPaper": "br",
36 "http://purl.org/spar/fabio/ReferenceBook": "br",
37 "http://purl.org/spar/fabio/Review": "br",
38 "http://purl.org/spar/fabio/ReviewArticle": "br",
39 "http://purl.org/spar/fabio/Series": "br",
40 "http://purl.org/spar/fabio/Thesis": "br",
41 "http://purl.org/spar/pro/RoleInTime": "ar",
42 "http://purl.org/spar/fabio/Manifestation": "re",
43 "http://xmlns.com/foaf/0.1/Agent": "ra",
44 "http://purl.org/spar/datacite/Identifier": "id",
45 }
47 def _process_entity_name(self, entity_name: str) -> tuple:
48 """
49 Process the entity name and format it for Redis storage.
51 :param entity_name: The entity name
52 :type entity_name: str
53 :return: A tuple containing the namespace and the processed entity name
54 :rtype: tuple
55 """
56 entity_name_str = str(entity_name)
57 if entity_name_str in self.entity_type_abbr:
58 return ("data", self.entity_type_abbr[entity_name_str])
59 else:
60 return ("prov", urllib.parse.quote(entity_name_str))
62 def set_counter(self, new_value: int, entity_name: str) -> None:
63 """
64 It allows to set the counter value of provenance entities.
66 :param new_value: The new counter value to be set
67 :type new_value: int
68 :param entity_name: The entity name
69 :type entity_name: str
70 :raises ValueError: if ``new_value`` is a negative integer.
71 :return: None
72 """
73 if new_value < 0:
74 raise ValueError("new_value must be a non negative integer!")
76 namespace, processed_entity_name = self._process_entity_name(entity_name)
77 key = f"{namespace}:{processed_entity_name}"
78 self.redis_client.set(key, new_value)
80 def read_counter(self, entity_name: str) -> int:
81 """
82 It allows to read the counter value of provenance entities.
84 :param entity_name: The entity name
85 :type entity_name: str
86 :return: The requested counter value.
87 """
88 namespace, processed_entity_name = self._process_entity_name(entity_name)
89 key = f"{namespace}:{processed_entity_name}"
90 result = self.redis_client.get(key)
92 if result:
93 return int(result)
94 else:
95 return 0
97 def increment_counter(self, entity_name: str) -> int:
98 """
99 It allows to increment the counter value of graph and provenance entities by one unit.
101 :param entity_name: The entity name
102 :type entity_name: str
103 :return: The newly-updated (already incremented) counter value.
104 """
105 namespace, processed_entity_name = self._process_entity_name(entity_name)
106 key = f"{namespace}:{processed_entity_name}"
107 new_count = self.redis_client.incr(key)
108 return new_count
110 def close(self):
111 """
112 Closes the Redis connection.
113 """
114 if self.redis_client:
115 self.redis_client.close()