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

1"""Convenience wrapper for the SQLite driver. 

2 

3This wrapper accepts a database path and optionally an "init" DDL script. 

4It executes queries given in a text form. 

5 

6Each query runs on its own connection, 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 is repeated on a new connection. 

12 

13""" 

14 

15import sqlite3 

16 

17import gws 

18 

19BUSY_TIMEOUT = 5.0 

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

21 

22MAX_ATTEMPTS = 3 

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

24 

25SLEEP_TIME = 0.1 

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

27 

28_RECOVERABLE_ERRORS = { 

29 'SQLITE_BUSY', 

30 'SQLITE_CANTOPEN', 

31 'SQLITE_LOCKED', 

32 'SQLITE_PROTOCOL', 

33} 

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

35 

36 

37class Error(gws.Error): 

38 pass 

39 

40 

41class Object: 

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

43 self.dbPath = db_path 

44 self.initDDL = init_ddl 

45 self.uidName = uid_column 

46 

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

48 """Execute a text DML statement.""" 

49 

50 self._exec2(False, stmt, params) 

51 

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

53 """Execute a text select statement.""" 

54 

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

56 

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

58 """Insert a new record (dict) into a table.""" 

59 

60 keys = ','.join(rec) 

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

62 

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

64 

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

66 """Update a record (dict) in a table.""" 

67 

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

69 self._exec2( 

70 False, 

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

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

73 ) 

74 

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

76 """Delete a record by uid from a table.""" 

77 

78 self._exec2( 

79 False, 

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

81 {'__uid': uid}, 

82 ) 

83 

84 ## 

85 

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

87 attempt = 0 

88 

89 while True: 

90 attempt += 1 

91 try: 

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

93 except sqlite3.Error as exc: 

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

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

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

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

98 gws.u.sleep(SLEEP_TIME) 

99 

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

101 conn = None 

102 

103 try: 

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

105 conn.row_factory = sqlite3.Row 

106 try: 

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

108 except sqlite3.OperationalError as exc: 

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

110 raise 

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

112 conn.executescript(self.initDDL) 

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

114 finally: 

115 if conn: 

116 conn.close() 

117 

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

119 cur = conn.execute(stmt, params) 

120 if is_select: 

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

122 return []