"""This module provide database manipulating toolkit.
Typical usage example:
.. code-block:: python
:linenos:
db = Database()
db.connect('some/path/to/sample.db')
db.affirmTable('foo')
db.setBool('bool_test', True, 'foo')
db.setObject('obj_test', {'hello':'world'}, 'foo')
obj = db.getObject('obj_test', 'foo')
print(obj)
# db.clear('foo')
db.close()
"""
import os, jsonpickle, sqlite3
from PySide6.QtCore import QByteArray, QDataStream
[docs]class Database:
"""Represent a database object powered by sqlite3.
Provide basic interface of database manipulating,
including a set of setters, getters, and other useful utils.
"""
[docs] def connect(self, path):
"""Load database of specified path.
Args:
str path: Source database path.
"""
self.conn = sqlite3.connect(path)
[docs] def close(self):
"""Close a connected database.
"""
self.conn.close()
del self.conn
#################### OPERATE ###################
[docs] def doSql(self, cmd, commit = True):
"""Execute specified sql command.
Args:
str cmd: Specified command to execute.
bool commit: Commit database operation.
"""
c = self.conn.cursor()
c.execute(cmd)
if commit: self.conn.commit()
c.close()
[docs] def createTable(self, table):
"""Create a database with specified name.
Args:
str table: Table name to create.
"""
self.doSql(f'''
CREATE TABLE {table} (
name STRING PRIMARY KEY
UNIQUE
NOT NULL,
data STRING
); ''')
[docs] def hasTable(self, table):
"""Check if has certain database table.
Args:
str table: Table name to check.
Returns:
bool: Contains certain table or not.
"""
c = self.conn.cursor()
c.execute(f'SELECT count(*) FROM sqlite_master WHERE type="table" AND name = "{table}"')
r = c.fetchone()
c.close()
return r[0] > 0
[docs] def affirmTable(self, table):
"""Affirm a certain database table.
If table not exist, will create a new one.
Args:
str table: Table name to affirm.
"""
if not self.hasTable(table):
self.createTable(table)
[docs] def dropTable(self, table):
"""Drop a certain database table.
Args:
str table: Table name to drop.
"""
self.doSql(f'DROP TABLE {table}')
[docs] def clearTable(self, table):
"""Clear a certain database table's content.
Args:
str table: Table name to clear.
"""
self.doSql(f'DELETE FROM {table}')
[docs] def remove(self, name, table):
"""Remove an entry in certain database table.
Args:
str name: Entry primary key to remove.
str table: Table name to operate.
"""
self.doSql(f'DELETE FROM {table} WHERE name == "{name}"')
#################### SETTERS ####################
[docs] def set(self, name, data, table):
"""Set an entry data in certain database table.
Args:
str name: Entry primary key to set.
any data: Data to set, will be implicit converted str.
str table: Table name to operate.
"""
if isinstance(data, str): data = f'"{data}"'
self.doSql(f'INSERT OR REPLACE INTO {table} VALUES ("{name}", {data});')
[docs] def setBool(self, name, data, table):
"""Set an boolean data in certain database table.
Args:
str name: Entry primary key to set.
bool data: Boolean data to set.
str table: Table name to operate.
"""
self.set(name, str(data), table)
[docs] def setQVariant(self, name, data, table):
"""Set an QVariant variable in certain database table.
Args:
str name: Entry primary key to set.
QVariant Data: QVariant data to set.
str table: Table name to operate.
"""
data = QByteArray()
stream = QDataStream(data, QIODevice.WriteOnly)
stream.writeQVariant(var)
data = str(data.toBase64(), 'utf-8')
self.set(name, f'b\'{data}\'', table)
[docs] def setObject(self, name, obj, table):
"""Set an object in certain database table.
Args:
str name: Entry primary key to set.
object obj: Object data to set, will be serialized into json string.
str table: Table name to operate.
"""
data = jsonpickle.encode(obj, indent = None)
self.set(name, data, table)
#################### GETTERS ####################
[docs] def get(self, name, table):
"""Get an entry data in certain database table.
Args:
str name: Entry primary key to get.
str table: Table name to get.
Returns:
str: Found entry data.
"""
c = self.conn.cursor()
c.execute(f'SELECT data FROM {table} WHERE name == "{name}"')
r = c.fetchone()
c.close()
return r and r[0]
[docs] def getBool(self, name, table):
"""Get an boolean data in certain database table.
Args:
str name: Fntry primary key to get.
str table: Fable name to get.
Returns:
bool: Found entry data.
"""
return self.get(name, table).lower() == 'true'
[docs] def getQVariant(self, name, table):
"""Get an QVariant data in certain database table.
Args:
str name: Entry primary key to get.
str table: Table name to get.
Returns:
QVariant: Found entry data.
"""
r = self.get(name, table)
if not r: return None
data = QByteArray.fromBase64(bytes(r[2:-1], 'utf-8'))
stream = QDataStream(data, QIODevice.ReadOnly)
return stream.readQVariant()
[docs] def getObject(self, name, table):
"""Get an object data in certain database table.
Args:
str name: Entry primary key to get.
str table: Table name to get.
Returns:
object: Found entry data.
"""
r = self.get(name, table)
if r: return jsonpickle.decode(r)
return None