Coverage for oc_ds_converter / oc_idmanager / oc_data_storage / sqlite_manager.py: 82%

1# SPDX-FileCopyrightText: 2023 Arianna Moretti <arianna.moretti4@unibo.it> 

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

3# 

4# SPDX-License-Identifier: ISC 

5 

6from __future__ import annotations 

7 

8import os.path 

9import pathlib 

10import sqlite3 

11 

12from oc_ds_converter.oc_idmanager.oc_data_storage.storage_manager import StorageManager 

13 

14 

15class SqliteStorageManager(StorageManager): 

16 """A concrete implementation of the ``StorageManager`` interface that persistently stores 

17 the IDs validity values within a SQLite database.""" 

18 

19 con: sqlite3.Connection 

20 cur: sqlite3.Cursor 

21 storage_filepath: str 

22 

23 def __init__(self, database: str | None = None, **params: object) -> None: 

24 """ 

25 Constructor of the ``SqliteStorageManager`` class. 

26 

27 :param database: The name of the database 

28 :type info_dir: str 

29 """ 

30 super().__init__(**params) 

31 sqlite3.threadsafety = 3 

32 if database and os.path.exists(database): 

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

34 self.storage_filepath = database 

35 elif database and not os.path.exists(database): 

36 if not os.path.exists(os.path.abspath(os.path.join(database, os.pardir))): 

37 pathlib.Path(os.path.abspath(os.path.join(database, os.pardir))).mkdir(parents=True, exist_ok=True) 

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

39 self.storage_filepath = database 

40 else: 

41 new_path_dir = os.path.join(os.getcwd(), "storage") 

42 if not os.path.exists(new_path_dir): 

43 os.makedirs(new_path_dir) 

44 new_path_db = os.path.join(new_path_dir, "id_valid_dict.db") 

45 self.con = sqlite3.connect(database=new_path_db) 

46 self.storage_filepath = new_path_db 

47 

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

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

50 id TEXT PRIMARY KEY, 

51 value INTEGER)""") 

52 

53 def set_full_value(self, id: str, value: dict[str, str | bool | object]) -> None: 

54 """ 

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

56 

57 :param value: The new counter value to be set 

58 :type value: dict 

59 :param id: The id string with prefix 

60 :type id: str 

61 :raises ValueError: if ``value`` is neither 0 nor 1 (0 is False, 1 is True). 

62 :return: None 

63 """ 

64 id_name = str(id) 

65 if not isinstance(value, dict): 

66 raise ValueError("value must be dict") 

67 if not isinstance(self.get_value(id_name), bool): 

68 valid = value.get("valid") 

69 if isinstance(valid, bool): 

70 self.set_value(id_name, valid) 

71 

72 def set_value(self, id: str, value: bool) -> None: 

73 """ 

74 It allows to set a value for the validity check of an id. 

75 

76 :param value: The new counter value to be set 

77 :type value: bool 

78 :param id: The id string with prefix 

79 :type id: str 

80 :raises ValueError: if ``value`` is neither 0 nor 1 (0 is False, 1 is True). 

81 :return: None 

82 """ 

83 id_name = str(id) 

84 if not isinstance(value, bool): 

85 raise ValueError("value must be boolean") 

86 validity = 1 if value else 0 

87 id_val = (id_name, validity) 

88 self.cur.execute("INSERT OR REPLACE INTO info VALUES (?,?)", id_val) 

89 self.con.commit() 

90 

91 def set_multi_value(self, list_of_tuples: list[tuple[str, bool]]) -> None: 

92 """ 

93 It allows to set a value for the validity check of an id. 

94 

95 :param value: The new counter value to be set 

96 :type value: bool 

97 :param id: The id string with prefix 

98 :type id: str 

99 :raises ValueError: if ``value`` is neither 0 nor 1 (0 is False, 1 is True). 

100 :return: None 

101 """ 

102 sqlite_list_copy: list[tuple[str, int]] = [] 

103 for t in list_of_tuples: 

104 if t[1] is True: 

105 sqlite_list_copy.append((t[0], 1)) 

106 else: 

107 sqlite_list_copy.append((t[0], 0)) 

108 self.cur.executemany("INSERT OR REPLACE INTO info VALUES (?,?)", sqlite_list_copy) 

109 self.con.commit() 

110 

111 def get_value(self, id: str) -> bool | None: 

112 """ 

113 It allows to read the value of the identifier. 

114 

115 :param id: The id name 

116 :type id: str 

117 :return: The requested id value (True if valid, False if invalid, None if not found). 

118 """ 

119 id_name = str(id) 

120 result = self.cur.execute(f"SELECT value FROM info WHERE id='{id_name}'") 

121 rows = result.fetchall() 

122 if len(rows) == 1: 

123 value = rows[0][0] 

124 return value == 1 

125 elif len(rows) == 0: 

126 return None 

127 else: 

128 raise Exception("There is more than one counter for this id. The database is broken") 

129 

130 def delete_storage(self) -> None: 

131 if os.path.exists(self.storage_filepath): 

132 try: 

133 self.con.close() 

134 os.remove(self.storage_filepath) 

135 except Exception: 

136 os.remove(self.storage_filepath) 

137 

138 def get_all_keys(self) -> list[str]: 

139 ids = [row[0] for row in self.cur.execute("SELECT id FROM info")] 

140 return ids