Coverage for oc_ocdm / counter_handler / sqlite_counter_handler.py: 88%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7# -*- coding: utf-8 -*- 

8 

9import sqlite3 

10 

11from oc_ocdm.counter_handler.counter_handler import CounterHandler 

12 

13 

14class SqliteCounterHandler(CounterHandler): 

15 """A concrete implementation of the ``CounterHandler`` interface that persistently stores 

16 the counter values within a SQLite database.""" 

17 

18 def __init__(self, database: str) -> None: 

19 """ 

20 Constructor of the ``SqliteCounterHandler`` class. 

21 

22 :param database: The name of the database 

23 :type info_dir: str 

24 """ 

25 sqlite3.threadsafety = 3 

26 self.database = database 

27 self.con = sqlite3.connect(database) 

28 self.cur = self.con.cursor() 

29 self.cur.execute("""CREATE TABLE IF NOT EXISTS info( 

30 entity TEXT PRIMARY KEY, 

31 count INTEGER)""") 

32 

33 def set_counter( 

34 self, 

35 new_value: int, 

36 entity_short_name: str, 

37 prov_short_name: str = "", # type: ignore[override] 

38 identifier: int = 1, 

39 supplier_prefix: str = "", 

40 ) -> None: 

41 """ 

42 It allows to set the counter value of provenance entities. 

43 

44 In this implementation, ``entity_short_name`` is used as a generic entity key 

45 (which may be a URI string when called from ProvSet with a GraphEntity). 

46 

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

48 :type new_value: int 

49 :param entity_short_name: The entity name (used as lookup key) 

50 :type entity_short_name: str 

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

52 :return: None 

53 """ 

54 if new_value < 0: 

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

56 self.cur.execute(f"INSERT OR REPLACE INTO info (entity, count) VALUES ('{entity_short_name}', {new_value})") 

57 self.con.commit() 

58 

59 def read_counter( 

60 self, 

61 entity_short_name: str, 

62 prov_short_name: str = "", # type: ignore[override] 

63 identifier: int = 1, 

64 supplier_prefix: str = "", 

65 ) -> int: 

66 """ 

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

68 

69 :param entity_short_name: The entity name (used as lookup key) 

70 :type entity_short_name: str 

71 :return: The requested counter value. 

72 """ 

73 rows = self.cur.execute(f"SELECT count FROM info WHERE entity='{entity_short_name}'").fetchall() 

74 if len(rows) == 1: 

75 return rows[0][0] 

76 elif len(rows) == 0: 

77 return 0 

78 else: 

79 raise (Exception("There is more than one counter for this entity. The databse id broken")) 

80 

81 def increment_counter( 

82 self, 

83 entity_short_name: str, 

84 prov_short_name: str = "", # type: ignore[override] 

85 identifier: int = 1, 

86 supplier_prefix: str = "", 

87 ) -> int: 

88 """ 

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

90 

91 :param entity_short_name: The entity name (used as lookup key) 

92 :type entity_short_name: str 

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

94 """ 

95 count = self.read_counter(entity_short_name) + 1 

96 self.set_counter(count, entity_short_name) 

97 return count 

98 

99 def increment_metadata_counter(self, entity_short_name: str = "", dataset_name: str = "") -> int: # type: ignore[override] 

100 return 0 

101 

102 def read_metadata_counter(self, entity_short_name: str = "", dataset_name: str = "") -> int: # type: ignore[override] 

103 return 0 

104 

105 def set_metadata_counter(self, new_value: int = 0, entity_short_name: str = "", dataset_name: str = "") -> None: # type: ignore[override] 

106 pass 

107 

108 def __getstate__(self): 

109 """ 

110 Support for pickle serialization. 

111 

112 Exclude the SQLite connection and cursor objects, which are not picklable. 

113 The database path is preserved and the connection will be recreated upon unpickling. 

114 """ 

115 state = self.__dict__.copy() 

116 del state["con"] 

117 del state["cur"] 

118 return state 

119 

120 def __setstate__(self, state: dict[str, object]) -> None: 

121 """ 

122 Support for pickle deserialization. 

123 

124 Recreates the SQLite connection and cursor after unpickling. 

125 """ 

126 vars(self).update(state) 

127 sqlite3.threadsafety = 3 

128 self.con = sqlite3.connect(self.database) 

129 self.cur = self.con.cursor()