Interface class for SQLite databases

Question

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')

Answer 1

Exception handling: you handle some exceptions but you print them to the console, where they can get lost/unnoticed. It would be better to use the logger module to record them to a file. What I like to do when writing classes is to attach to the logger routine of the main module. This way, all messages go to the console AND get written to the same log file
The code would be along these lines:

import logging

def __init__(self, file, logger=None):
    self.dbfile = file

    # acquire existing logger - https://fangpenlin.com/posts/2012/08/26/good-logging-practice-in-python/
    self.logger = logger or logging.getLogger(__name__)
    self.logger.debug('Running __init__')

See the section 'Use name as the logger name' in the link above.

---

Redundant code:

assert os.path.exists(self.dbfile)
assert os.path.isfile(self.dbfile)

isfile is sufficient.

os.path.isfile(path)

Return True if path is an existing regular file. This follows symbolic links, so both islink() and isfile() can be true for the same path.

Source: os.path.isfile(path)

But your code will fail if the DB file does not exist. By default SQLite connect will create the file if it doesn't already exist. So I think you should preferably make that check after the call to connect to determine that the file is indeed present.

---

Return codes: your procedures return either 0 or -1, I wonder why you made that arbitrary choice. You could simply return a boolean value, also in your except blocks if you want, so that there is a return value on all paths.

Your read procedure returns a tuple (note that fetchall returns a list):

return 0, rows;

Personally I would only return the rows, but in case of error I would return None instead, so I know that the operation failed. If you don't look at the error code, the (empty) returned list is still iterable.
The return code doesn't matter here because it doesn't give any additional information. If I need details, I will look at the error log.

---

I would make the call to close optional (it's going to be forgotten sometimes anyway). Simply register a cleanup routine for your class:

import atexit

def __init__(self, iface, logger=None):
    ...
    # run cleanup routine when class terminates
    atexit.register(self.cleanup)

def cleanup(self):
    """To perform cleanup tasks when class instance is shutting down
    """
    self.logger.debug('Closing DB')
    if hasattr(self, '_dbconn'):
        self._dbconn.close()

And in fact the open method is not required either:

res = db.open()

Just move the code to the __init__ function. No need to do open & close.

---

Your write routine only allows certain operations:

assert any(map(lambda s: query.upper().strip().startswith(s), ('CREATE TABLE', 'ALTER TABLE', 'INSERT', 'UPDATE', 'DELETE')))

Therefore no PRAGMAs allowed but this is your choice.

One possible enhancement would be to take an existing connection as a class argument, in case you want to do something special on an open DB (like a PRAGMA), that your class does not easily allow. Although I can imagine this code is sufficient for your needs.