Coverage for gws-app/gws/lib/sqlitex/__init__.py: 99%
70 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-23 20:06 +0200
1"""Convenience wrapper for the SQLite driver.
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``).
6Each query runs on its own connection in autocommit mode, which is closed immediately afterwards.
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.
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``.
14Example::
16 import gws.lib.sqlitex
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"""
27import sqlite3
29import gws
31BUSY_TIMEOUT = 5.0
32"""Time in seconds to wait for a lock before giving up."""
34MAX_ATTEMPTS = 3
35"""How many times to repeat a query that failed with a recoverable error."""
37SLEEP_TIME = 0.1
38"""Time in seconds to wait between the attempts."""
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."""
49class Error(gws.Error):
50 """Raised when a query fails."""
52 pass
55class Object:
56 """SQLite database wrapper."""
58 def __init__(self, db_path: str, init_ddl: str = '', uid_column: str = 'uid'):
59 """Create a database wrapper.
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
70 def execute(self, stmt: str, **params):
71 """Execute a text statement that does not return rows.
73 Args:
74 stmt: SQL statement.
75 **params: Values for the named parameters of the statement.
77 Raises:
78 ``Error``: If the statement fails.
79 """
81 self._exec2(False, stmt, params)
83 def select(self, stmt: str, **params) -> list[dict]:
84 """Execute a text select statement.
86 Args:
87 stmt: SQL statement.
88 **params: Values for the named parameters of the statement.
90 Returns:
91 Result rows as dicts.
93 Raises:
94 ``Error``: If the statement fails.
95 """
97 return self._exec2(True, stmt, params)
99 def insert(self, table_name: str, rec: dict, on_conflict: str = ''):
100 """Insert a record into a table.
102 Args:
103 table_name: Table name.
104 rec: Record as a dict of column names and values.
105 on_conflict: What to do if the record conflicts with an existing primary or unique key:
106 empty to fail, ``ignore`` to skip the record, ``update`` to update the existing record
107 with the given values (the conflict is detected on the primary key column).
109 Raises:
110 ``Error``: If the statement fails, or ``on_conflict`` is invalid.
111 """
113 keys = ','.join(rec)
114 vals = ','.join(':' + k for k in rec)
116 if not on_conflict:
117 clause = ''
118 elif on_conflict == 'ignore':
119 clause = ' ON CONFLICT DO NOTHING'
120 elif on_conflict == 'update':
121 sets = ','.join(f'{k}=excluded.{k}' for k in rec if k != self.uidName)
122 clause = f' ON CONFLICT({self.uidName}) DO UPDATE SET {sets}'
123 else:
124 raise Error(f'sqlitex: invalid {on_conflict=}')
126 self._exec2(False, f'INSERT INTO {table_name} ({keys}) VALUES({vals}){clause}', rec)
128 def update(self, table_name: str, rec: dict, uid):
129 """Update a record in a table.
131 Args:
132 table_name: Table name.
133 rec: Columns to update, as a dict of column names and values.
134 uid: Value of the primary key column of the record.
136 Raises:
137 ``Error``: If the statement fails.
138 """
140 vals = ','.join(f'{k}=:{k}' for k in rec)
141 self._exec2(
142 False,
143 f'UPDATE {table_name} SET {vals} WHERE {self.uidName}=:__uid',
144 {'__uid': uid, **rec},
145 )
147 def delete(self, table_name: str, uid):
148 """Delete a record from a table.
150 Args:
151 table_name: Table name.
152 uid: Value of the primary key column of the record.
154 Raises:
155 ``Error``: If the statement fails.
156 """
158 self._exec2(
159 False,
160 f'DELETE FROM {table_name} WHERE {self.uidName}=:__uid',
161 {'__uid': uid},
162 )
164 ##
166 def _exec2(self, is_select, stmt, params):
167 attempt = 0
169 while True:
170 attempt += 1
171 try:
172 return self._exec3(is_select, stmt, params)
173 except sqlite3.Error as exc:
174 gws.log.warning(f'sqlitex: {self.dbPath}: {exc}, sql={" ".join(stmt.split())}')
175 name = getattr(exc, 'sqlite_errorname', '')
176 if not any(name.startswith(e) for e in _RECOVERABLE_ERRORS) or attempt >= MAX_ATTEMPTS:
177 raise Error(f'sqlitex: {self.dbPath}: {exc}') from exc
178 gws.u.sleep(SLEEP_TIME)
180 def _exec3(self, is_select, stmt, params):
181 conn = None
183 try:
184 conn = sqlite3.connect(self.dbPath, timeout=BUSY_TIMEOUT, isolation_level=None)
185 conn.row_factory = sqlite3.Row
186 try:
187 return self._exec4(conn, is_select, stmt, params)
188 except sqlite3.OperationalError as exc:
189 if not self.initDDL or 'no such table' not in str(exc):
190 raise
191 gws.log.warning(f'sqlitex: {self.dbPath}: {exc}, running init...')
192 conn.executescript(self.initDDL)
193 return self._exec4(conn, is_select, stmt, params)
194 finally:
195 if conn:
196 conn.close()
198 def _exec4(self, conn, is_select, stmt, params):
199 cur = conn.execute(stmt, params)
200 if is_select:
201 return [dict(r) for r in cur]
202 return []