Database logging and error handling
This commit is contained in:
+96
-63
@@ -5,98 +5,131 @@ same code can run on CPython (`sqlite3`) and MicroPython (`sqlite3` or
|
||||
`usqlite`, depending on the port).
|
||||
"""
|
||||
|
||||
from shared.logging import log
|
||||
|
||||
try:
|
||||
import sqlite3 as _sqlite
|
||||
DRIVER_NAME = "sqlite3"
|
||||
import sqlite3 as _sqlite
|
||||
DRIVER_NAME = "sqlite3"
|
||||
except ImportError:
|
||||
try:
|
||||
import usqlite as _sqlite
|
||||
DRIVER_NAME = "usqlite"
|
||||
except ImportError as exc:
|
||||
raise ImportError("No sqlite driver found. Expected sqlite3 or usqlite.") from exc
|
||||
try:
|
||||
import usqlite as _sqlite
|
||||
DRIVER_NAME = "usqlite"
|
||||
except ImportError as exc:
|
||||
log("[DB Error] No sqlite driver found. Expected sqlite3 or usqlite.")
|
||||
raise ImportError("No sqlite driver found. Expected sqlite3 or usqlite.") from exc
|
||||
|
||||
|
||||
def _connect(database_path, **connect_kwargs):
|
||||
if connect_kwargs:
|
||||
try:
|
||||
return _sqlite.connect(database_path, **connect_kwargs)
|
||||
except TypeError:
|
||||
pass
|
||||
return _sqlite.connect(database_path)
|
||||
try:
|
||||
if connect_kwargs:
|
||||
try:
|
||||
return _sqlite.connect(database_path, **connect_kwargs)
|
||||
except TypeError:
|
||||
pass
|
||||
return _sqlite.connect(database_path)
|
||||
except Exception as e:
|
||||
log(f"[DB Error] Driver connect failed for '{database_path}': {e}")
|
||||
raise
|
||||
|
||||
|
||||
class Database:
|
||||
"""Lightweight connection wrapper with a consistent API."""
|
||||
"""Lightweight connection wrapper with logging and consistent API."""
|
||||
|
||||
def __init__(self, database_path, **connect_kwargs):
|
||||
self._database_path = database_path
|
||||
self._connect_kwargs = connect_kwargs
|
||||
self._connection = None
|
||||
def __init__(self, database_path, **connect_kwargs):
|
||||
self._database_path = database_path
|
||||
self._connect_kwargs = connect_kwargs
|
||||
self._connection = None
|
||||
|
||||
def open(self):
|
||||
if self._connection is None:
|
||||
self._connection = _connect(self._database_path, **self._connect_kwargs)
|
||||
return self._connection
|
||||
def open(self):
|
||||
if self._connection is None:
|
||||
self._connection = _connect(self._database_path, **self._connect_kwargs)
|
||||
return self._connection
|
||||
|
||||
def close(self):
|
||||
if self._connection is not None:
|
||||
self._connection.close()
|
||||
self._connection = None
|
||||
def close(self):
|
||||
if self._connection is not None:
|
||||
try:
|
||||
self._connection.close()
|
||||
except Exception as e:
|
||||
log(f"[DB Error] Failed to close database '{self._database_path}': {e}")
|
||||
finally:
|
||||
self._connection = None
|
||||
|
||||
def commit(self):
|
||||
connection = self.open()
|
||||
if hasattr(connection, "commit"):
|
||||
connection.commit()
|
||||
def commit(self):
|
||||
connection = self.open()
|
||||
if hasattr(connection, "commit"):
|
||||
try:
|
||||
connection.commit()
|
||||
except Exception as e:
|
||||
log(f"[DB Error] Commit failed on '{self._database_path}': {e}")
|
||||
raise
|
||||
|
||||
def cursor(self):
|
||||
return self.open().cursor()
|
||||
def cursor(self):
|
||||
return self.open().cursor()
|
||||
|
||||
def execute(self, sql, params=None):
|
||||
cursor = self.cursor()
|
||||
if params is None:
|
||||
cursor.execute(sql)
|
||||
else:
|
||||
cursor.execute(sql, params)
|
||||
return cursor
|
||||
def execute(self, sql, params=None):
|
||||
cursor = self.cursor()
|
||||
try:
|
||||
if params is None:
|
||||
cursor.execute(sql)
|
||||
else:
|
||||
cursor.execute(sql, params)
|
||||
return cursor
|
||||
except Exception as e:
|
||||
log(f"[DB Error] Query failed on '{self._database_path}' | SQL: {sql} | Params: {params} | Error: {e}")
|
||||
raise
|
||||
|
||||
def executemany(self, sql, params_list):
|
||||
cursor = self.cursor()
|
||||
cursor.executemany(sql, params_list)
|
||||
return cursor
|
||||
def executemany(self, sql, params_list):
|
||||
cursor = self.cursor()
|
||||
try:
|
||||
cursor.executemany(sql, params_list)
|
||||
return cursor
|
||||
except Exception as e:
|
||||
log(f"[DB Error] Executemany failed on '{self._database_path}' | SQL: {sql} | Error: {e}")
|
||||
raise
|
||||
|
||||
def fetchone(self, sql, params=None):
|
||||
return self.execute(sql, params).fetchone()
|
||||
def fetchone(self, sql, params=None):
|
||||
return self.execute(sql, params).fetchone()
|
||||
|
||||
def fetchall(self, sql, params=None):
|
||||
return self.execute(sql, params).fetchall()
|
||||
def fetchall(self, sql, params=None):
|
||||
return self.execute(sql, params).fetchall()
|
||||
|
||||
def executescript(self, script):
|
||||
connection = self.open()
|
||||
if hasattr(connection, "executescript"):
|
||||
return connection.executescript(script)
|
||||
raise NotImplementedError("executescript is not available on this sqlite backend")
|
||||
def executescript(self, script):
|
||||
connection = self.open()
|
||||
if hasattr(connection, "executescript"):
|
||||
try:
|
||||
return connection.executescript(script)
|
||||
except Exception as e:
|
||||
log(f"[DB Error] Executescript failed on '{self._database_path}': {e}")
|
||||
raise
|
||||
log("[DB Error] executescript is not available on this sqlite backend")
|
||||
raise NotImplementedError("executescript is not available on this sqlite backend")
|
||||
|
||||
def __enter__(self):
|
||||
self.open()
|
||||
return self
|
||||
def __enter__(self):
|
||||
self.open()
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, traceback):
|
||||
if exc_type is None:
|
||||
self.commit()
|
||||
self.close()
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
if exc_type is None:
|
||||
self.commit()
|
||||
else:
|
||||
log(f"[DB Error] Context exited with exception on '{self._database_path}': {exc_val}")
|
||||
self.close()
|
||||
|
||||
|
||||
def connect(database_path, **connect_kwargs):
|
||||
return Database(database_path, **connect_kwargs)
|
||||
return Database(database_path, **connect_kwargs)
|
||||
|
||||
|
||||
def execute(database_path, sql, params=None, **connect_kwargs):
|
||||
return connect(database_path, **connect_kwargs).execute(sql, params)
|
||||
with connect(database_path, **connect_kwargs) as db_inst:
|
||||
return db_inst.execute(sql, params)
|
||||
|
||||
|
||||
def fetchone(database_path, sql, params=None, **connect_kwargs):
|
||||
return connect(database_path, **connect_kwargs).fetchone(sql, params)
|
||||
with connect(database_path, **connect_kwargs) as db_inst:
|
||||
return db_inst.fetchone(sql, params)
|
||||
|
||||
|
||||
def fetchall(database_path, sql, params=None, **connect_kwargs):
|
||||
return connect(database_path, **connect_kwargs).fetchall(sql, params)
|
||||
with connect(database_path, **connect_kwargs) as db_inst:
|
||||
return db_inst.fetchall(sql, params)
|
||||
Reference in New Issue
Block a user