103 lines
2.7 KiB
Python
103 lines
2.7 KiB
Python
"""Small sqlite compatibility layer for CPython and MicroPython.
|
|
|
|
The module exposes a tiny wrapper around the available sqlite driver so the
|
|
same code can run on CPython (`sqlite3`) and MicroPython (`sqlite3` or
|
|
`usqlite`, depending on the port).
|
|
"""
|
|
|
|
try:
|
|
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
|
|
|
|
|
|
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)
|
|
|
|
|
|
class Database:
|
|
"""Lightweight connection wrapper with a consistent API."""
|
|
|
|
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 close(self):
|
|
if self._connection is not None:
|
|
self._connection.close()
|
|
self._connection = None
|
|
|
|
def commit(self):
|
|
connection = self.open()
|
|
if hasattr(connection, "commit"):
|
|
connection.commit()
|
|
|
|
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 executemany(self, sql, params_list):
|
|
cursor = self.cursor()
|
|
cursor.executemany(sql, params_list)
|
|
return cursor
|
|
|
|
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 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 __enter__(self):
|
|
self.open()
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, traceback):
|
|
if exc_type is None:
|
|
self.commit()
|
|
self.close()
|
|
|
|
|
|
def connect(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)
|
|
|
|
|
|
def fetchone(database_path, sql, params=None, **connect_kwargs):
|
|
return connect(database_path, **connect_kwargs).fetchone(sql, params)
|
|
|
|
|
|
def fetchall(database_path, sql, params=None, **connect_kwargs):
|
|
return connect(database_path, **connect_kwargs).fetchall(sql, params)
|