Coverage for rdflib_ocdm / counter_handler / in_memory_counter_handler.py: 100%

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2016 Silvio Peroni <essepuntato@gmail.com> 

4# SPDX-FileCopyrightText: 2023-2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

5# 

6# SPDX-License-Identifier: ISC 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING 

11 

12if TYPE_CHECKING: 

13 from typing import List, Dict 

14 

15from rdflib_ocdm.counter_handler.counter_handler import CounterHandler 

16 

17 

18class InMemoryCounterHandler(CounterHandler): 

19 """A concrete implementation of the ``CounterHandler`` interface that temporarily stores 

20 the counter values in the volatile system memory.""" 

21 

22 def __init__(self) -> None: 

23 """ 

24 Constructor of the ``InMemoryCounterHandler`` class. 

25 """ 

26 self.prov_counters = dict() 

27 

28 def set_counter(self, new_value: int, entity_name: str) -> None: 

29 """ 

30 It allows to set the counter value of graph and provenance entities. 

31 

32 :param new_value: The new counter value to be set 

33 :type new_value: int 

34 :param entity_name: The entity name 

35 :type entity_name: str 

36 :raises ValueError: if ``new_value`` is a negative integer. 

37 :return: None 

38 """ 

39 entity_name = str(entity_name) 

40 if new_value < 0: 

41 raise ValueError("new_value must be a non negative integer!") 

42 self.prov_counters[entity_name] = new_value 

43 

44 def read_counter(self, entity_name: str) -> int: 

45 """ 

46 It allows to read the counter value of provenance entities. 

47 

48 :param entity_name: The entity name 

49 :type entity_name: str 

50 :return: The requested counter value. 

51 """ 

52 entity_name = str(entity_name) 

53 if entity_name in self.prov_counters: 

54 return self.prov_counters[entity_name] 

55 else: 

56 self.prov_counters[entity_name] = 0 

57 return 0 

58 

59 def increment_counter(self, entity_name: str) -> int: 

60 """ 

61 It allows to increment the counter value of graph and provenance entities by one unit. 

62 

63 :param entity_name: The entity name 

64 :type entity_name: str 

65 :return: The newly-updated (already incremented) counter value. 

66 """ 

67 entity_name = str(entity_name) 

68 if entity_name in self.prov_counters: 

69 self.prov_counters[entity_name] += 1 

70 else: 

71 self.prov_counters[entity_name] = 1 

72 return self.prov_counters[entity_name]