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-2026 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 pass 

14 

15from rdflib_ocdm.counter_handler.counter_handler import CounterHandler 

16 

17 

18class InMemoryCounterHandler(CounterHandler): 

19 """A concrete implementation of the ``CounterHandler`` interface 

20 that temporarily stores the counter values in the volatile system 

21 memory.""" 

22 

23 def __init__(self) -> None: 

24 """ 

25 Constructor of the ``InMemoryCounterHandler`` class. 

26 """ 

27 self.prov_counters = dict() 

28 

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

30 """ 

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

32 

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

34 :type new_value: int 

35 :param entity_name: The entity name 

36 :type entity_name: str 

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

38 :return: None 

39 """ 

40 entity_name = str(entity_name) 

41 if new_value < 0: 

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

43 self.prov_counters[entity_name] = new_value 

44 

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

46 """ 

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

48 

49 :param entity_name: The entity name 

50 :type entity_name: str 

51 :return: The requested counter value. 

52 """ 

53 entity_name = str(entity_name) 

54 if entity_name in self.prov_counters: 

55 return self.prov_counters[entity_name] 

56 else: 

57 self.prov_counters[entity_name] = 0 

58 return 0 

59 

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

61 """ 

62 It allows to increment the counter value of graph and 

63 provenance entities by one unit. 

64 

65 :param entity_name: The entity name 

66 :type entity_name: str 

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

68 """ 

69 entity_name = str(entity_name) 

70 if entity_name in self.prov_counters: 

71 self.prov_counters[entity_name] += 1 

72 else: 

73 self.prov_counters[entity_name] = 1 

74 return self.prov_counters[entity_name]