Coverage for gws-app/gws/lib/sqlitex/__init__.py: 98%

1"""Convenience wrapper for the SQLite driver. 

2 

3The ``Object`` class accepts a database path and optionally an "init" DDL script. 

4It executes queries given in a text form, with named parameters (``:name``). 

5 

6Each query runs on its own connection in autocommit mode, which is closed immediately afterwards. 

7 

8If a query fails with "no such table", the wrapper runs the "init" script and repeats the query once. 

9The script can contain multiple statements. 

10 

11A query that fails with a recoverable error (e.g. the database is locked) is repeated on a new connection, 

12up to ``MAX_ATTEMPTS`` times. Other errors are raised as ``Error``. 

13 

14Example:: 

15 

16 import gws.lib.sqlitex 

17 

18 db = gws.lib.sqlitex.Object( 

19 '/data/jobs.sqlite', 

20 init_ddl='CREATE TABLE jobs (uid TEXT PRIMARY KEY, state TEXT)', 

21 ) 

22 db.insert('jobs', {'uid': 'a1', 'state': 'open'}) 

23 db.update('jobs', {'state': 'done'}, 'a1') 

24 rows = db.select('SELECT * FROM jobs WHERE state=:state', state='done') 

25""" 

26 

27import sqlite3 

28 

29import gws 

30 

31BUSY_TIMEOUT = 5.0 

32"""Time in seconds to wait for a lock before giving up.""" 

33 

34MAX_ATTEMPTS = 3 

35"""How many times to repeat a query that failed with a recoverable error.""" 

36 

37SLEEP_TIME = 0.1 

38"""Time in seconds to wait between the attempts.""" 

39 

40_RECOVERABLE_ERRORS = { 

41 'SQLITE_BUSY', 

42 'SQLITE_CANTOPEN', 

43 'SQLITE_LOCKED', 

44 'SQLITE_PROTOCOL', 

45} 

46"""Errors worth repeating, matched against the leading part of the sqlite error name.""" 

47 

48 

49class Error(gws.Error): 

50 """Raised when a query fails.""" 

51 

52 pass 

53 

54 

55class Object: 

56 """SQLite database wrapper.""" 

57 

58 def __init__(self, db_path: str, init_ddl: str = '', uid_column: str = 'uid'): 

59 """Create a database wrapper. 

60 

61 Args: 

62 db_path: Path to the database file. 

63 init_ddl: DDL script to run when a table does not exist. 

64 uid_column: Name of the primary key column, used by ``update`` and ``delete``. 

65 """ 

66 self.dbPath = db_path 

67 self.initDDL = init_ddl 

68 self.uidName = uid_column 

69 

70 def execute(self, stmt: str, **params): 

71 """Execute a text statement that does not return rows. 

72 

73 Args: 

74 stmt: SQL statement. 

75 **params: Values for the named parameters of the statement. 

76 

77 Raises: 

78 ``Error``: If the statement fails. 

79 """ 

80 

81 self._exec2(False, stmt, params) 

82 

83 def select(self, stmt: str, **params) -> list[dict]: 

84 """Execute a text select statement. 

85 

86 Args: 

87 stmt: SQL statement. 

88 **params: Values for the named parameters of the statement. 

89 

90 Returns: 

91 Result rows as dicts. 

92 

93 Raises: 

94 ``Error``: If the statement fails. 

95 """ 

96 

97 return self._exec2(True, stmt, params) 

98 

99 def insert(self, table_name: str, rec: dict): 

100 """Insert a record into a table. 

101 

102 Args: 

103 table_name: Table name. 

104 rec: Record as a dict of column names and values. 

105 

106 Raises: 

107 ``Error``: If the statement fails. 

108 """ 

109 

110 keys = ','.join(rec) 

111 vals = ','.join(':' + k for k in rec) 

112 

113 self._exec2(False, f'INSERT INTO {table_name} ({keys}) VALUES({vals})', rec) 

114 

115 def update(self, table_name: str, rec: dict, uid): 

116 """Update a record in a table. 

117 

118 Args: 

119 table_name: Table name. 

120 rec: Columns to update, as a dict of column names and values. 

121 uid: Value of the primary key column of the record. 

122 

123 Raises: 

124 ``Error``: If the statement fails. 

125 """ 

126 

127 vals = ','.join(f'{k}=:{k}' for k in rec) 

128 self._exec2( 

129 False, 

130 f'UPDATE {table_name} SET {vals} WHERE {self.uidName}=:__uid', 

131 {'__uid': uid, **rec}, 

132 ) 

133 

134 def delete(self, table_name: str, uid): 

135 """Delete a record from a table. 

136 

137 Args: 

138 table_name: Table name. 

139 uid: Value of the primary key column of the record. 

140 

141 Raises: 

142 ``Error``: If the statement fails. 

143 """ 

144 

145 self._exec2( 

146 False, 

147 f'DELETE FROM {table_name} WHERE {self.uidName}=:__uid', 

148 {'__uid': uid}, 

149 ) 

150 

151 ## 

152 

153 def _exec2(self, is_select, stmt, params): 

154 attempt = 0 

155 

156 while True: 

157 attempt += 1 

158 try: 

159 return self._exec3(is_select, stmt, params) 

160 except sqlite3.Error as exc: 

161 gws.log.warning(f'sqlitex: {self.dbPath}: {exc}, sql={" ".join(stmt.split())}') 

162 name = getattr(exc, 'sqlite_errorname', '') 

163 if not any(name.startswith(e) for e in _RECOVERABLE_ERRORS) or attempt >= MAX_ATTEMPTS: 

164 raise Error(f'sqlitex: {self.dbPath}: {exc}') from exc 

165 gws.u.sleep(SLEEP_TIME) 

166 

167 def _exec3(self, is_select, stmt, params): 

168 conn = None 

169 

170 try: 

171 conn = sqlite3.connect(self.dbPath, timeout=BUSY_TIMEOUT, isolation_level=None) 

172 conn.row_factory = sqlite3.Row 

173 try: 

174 return self._exec4(conn, is_select, stmt, params) 

175 except sqlite3.OperationalError as exc: 

176 if not self.initDDL or 'no such table' not in str(exc): 

177 raise 

178 gws.log.warning(f'sqlitex: {self.dbPath}: {exc}, running init...') 

179 conn.executescript(self.initDDL) 

180 return self._exec4(conn, is_select, stmt, params) 

181 finally: 

182 if conn: 

183 conn.close() 

184 

185 def _exec4(self, conn, is_select, stmt, params): 

186 cur = conn.execute(stmt, params) 

187 if is_select: 

188 return [dict(r) for r in cur] 

189 return []