I am learning the basics about managing an SQLite database by using Python and, mainly for practice purposes, I have then defined an interface class in order to let operations on database as clear as possible. That said, I would like to ask you for some feedback about the following code:
import sqlite3, os
class DbInterface:
def __init__(self, file):
self.dbfile = file
def open(self):
try:
assert os.path.exists(self.dbfile)
assert os.path.isfile(self.dbfile)
self._dbconn = sqlite3.connect(self.dbfile)
return 0
except sqlite3.Error as e:
print(e)
except AssertionError as e:
print('ERROR: database file not found.')
return -1
def close(self):
if hasattr(self, '_dbconn'):
self._dbconn.close()
return 0
return -1
def write(self, query, *args):
try:
assert any(map(lambda s: query.upper().strip().startswith(s), ('CREATE TABLE', 'ALTER TABLE', 'INSERT', 'UPDATE', 'DELETE')))
assert query.count('?') == len(args)
c = self._dbconn.cursor()
c.execute(query, args)
self._dbconn.commit()
return 0
except AssertionError:
print('ERROR: inconsistent query arguments.')
except sqlite3.Error as e:
self._dbconn.rollback()
print('SQL ERROR:', e)
return -1
def read(self, query, *args):
try:
assert query.count('?') == len(args)
assert query.upper().strip().startswith('SELECT')
c = self._dbconn.cursor()
c.execute(query, args)
rows = c.fetchall()
return 0, rows;
except AssertionError:
print('ERROR: inconsistent query arguments.')
except sqlite3.Error as e:
print('SQL ERROR:', e)
return -1, ()
Here is a test function I am using for code validation:
def test(path):
db = DbInterface(path)
res = db.open()
q = """
CREATE TABLE IF NOT EXISTS example(
id integer PRIMARY KEY,
sometext text NOT NULL
);
"""
res = db.write(q)
q = 'INSERT INTO example(sometext) VALUES(?);'
res = db.write(q, 'a')
q = 'INSERT INTO example(sometext) VALUES(?, ?);'
res = db.write(q, 'b')
q = 'INSERT INTO example(sometext) VALUES(?);'
res = db.write(q)
q = 'INSERT INTO example(sometext) VALUES(?, ?);'
res = db.write(q, 'c', 'd')
q = 'SELECT * FROM example;'
res, data = db.read(q)
q = 'SELECT somevalue FROM example;'
res, data = db.read(q)
q = 'SELECT * FROM example WHERE sometext = ?;'
res, data = db.read(q, 'a')
q = 'SELECT * FROM example WHERE sometext = ?;'
res, data = db.read(q)
res = db.close()
if __name__ == '__main__':
test('testdb.db')