Database abstraction class

Question

The class does some database abstraction operations. And while I could have just used an ORM, I prefer to get my hands dirty when learning something new. There are no bugs (not as far as I know anyway) and the code is working as intended. But, since I'm learning Python, I want to make sure I'm thinking and working in the most "pythonic" way possible.

from Singleton import Singleton;

#Singleton class implements static methods and variable required for the
#singleton design pattern
class Database(Singleton):

    def __init__(self, **kwargs):

        self._con = None;
        self._host = kwargs['host'] if kwargs.has_key('host') else 'localhost';
        self._username = kwargs['username'];
        self._password = kwargs['password'];
        self._dbname = kwargs['database'];
        self._driver = kwargs['driver'];

        #encloser characters for system identifiers(key_delim) and strings(str_delim)
        self._key_delim = '"';
        self._str_delim = "'";

        if self._driver == 'mysql':
            self._module = __import__('MySQLdb');
            self._key_delim = '`';
        elif self._driver == 'pgsql':
            self._module = __import__('psycopg2');
        else:
            raise Exception("Unknown database driver");

        self._affected_rows = None;
        self._last_query = None;
        self._insert_id = None;
        self._error = None;
        self._autocommit = False;

        self.connect();

    def __del__(self):
        self.disconnect();

    def connect(self):
        kwargs = {'host': self._host, 'user': self._username};
        if self._driver == 'mysql':
            kwargs['passwd'] = self._password;
            kwargs['db'] = self._dbname;
        elif self._driver == 'pgsql':
            kwargs['database'] = self._dbname;
            kwargs['password'] = self._password;

        self._con = self._module.connect(**kwargs);

    def disconnect(self):
        if self._con:
            self._con.commit();
            self._con.close();

    def reconnect(self):
        self.disconnect();
        self.connect();

    #fake connecting to the database. Useful when trying out connection parameters, e.g. during install
    @staticmethod
    def mock(**kwargs):
        try:
            d = Database(kwargs);
            return true;
        except Exception:
            return false;


    #queries the database, returning the result as a list of dicts or None, if no row found or on commands
    def _query(self, s, params = None):

        if isinstance(params, list):
            params = tuple(params);

        #need this for compatibility with manual queries using MySQL format, where the backtick is used for enclosing column names
        #instead of the standard double quote. Will be removed soon
        if self._driver != 'mysql':
            s = self.__replace_backticks(s);

        try:
            cur = self._con.cursor();
            cur.execute(s, params);
            self._insert_id = cur.lastrowid;
            self._affected_rows = cur.rowcount;

            try:
                results = cur.fetchall();
                n = len(results);
                cols = self.table_columns(None, cur);
            except self._module.DatabaseError:
                #INSERT/UPDATE or similar
                return None;
            finally:
                cur.close();

            retval = [];
            for i in range(0,n):
                aux = results[i];
                row = {};
                for j in range(0,len(cols)):
                    #elem = aux[j].decode('UTF-8') if isinstance(aux[j], basestring) else aux[j];
                    row[cols[j]] = aux[j];

                if len(row):
                    retval.append(row);

            return retval;

        except self._module.DatabaseError as e:
            #Error. Reset insert id and affected rows to None
            self._insert_id = None;
            self._affected_rows = None;
            raise Exception("Database Error: %s" % str(e));

        return retval;

    #escape a variable/tuple/list
    def escape(self, s):
        if isinstance(s, basestring):
            return self._con.escape_string(s);
        elif isinstance(s, list):
            return map(lambda x: self.escape(x), s);
        elif isinstance(s, tuple):
            return tuple(self.escape(list(s)));
        else:
            raise TypeException("Unknown parameter given for escaping");

        #never get here
        return None;

    #encloses a string with single quotes
    def enclose_str(self, s):
        if isinstance(s, basestring):
            return ''.join([self._str_delim,str(s),self._str_delim]);
        elif isinstance(s, list):
            return map(self.enclose_str, s);
        elif isinstance(s, tuple):
            return tuple(map(self.enclose_str, s));
        else:
            raise TypeError("Unknown argument type to enclose_str");

    #encloses an identifier in the appropriate double quotes/backticks  
    def enclose_sys(self,s):
        #we do not enclose variable containing spaces because we assume them to be expressions, e.g. COUNT(*) AS ...
        #Column names containing spaces are not supported
        if isinstance(s, basestring):
            if s.count(' ') or s == '*':
                return s;
            return ''.join([self._key_delim,str(s),self._key_delim]);
        elif isinstance(s, list):
            return map(self.enclose_sys, s);
        elif isinstance(s, tuple):
            return tuple(map(self.enclose_sys, s));
        else:
            raise TypeError("Unknown argument type to enclose_sys");

    #SELECT FROM table
    def select(self, table, columns = None, where = None, op = "AND"):
        if isinstance(columns, tuple):
            columns = ",".join(map(lambda x: self.enclose_sys(x), columns));
        elif isinstance(columns, basestring):
            columns = self.enclose_sys(columns);
        elif not columns:
            columns = "*";
        else:
            raise TypeException("Invalid column definition");

        (where_clause, where_params) = self.__expand_where_clause(where, op);

        if not where_clause:
            return self._query("SELECT %s FROM %s" % (columns, self.enclose_sys(table)));
        else:
            return self._query("SELECT %s FROM %s WHERE %s" % (columns, self.enclose_sys(table), where_clause), where_params);

    #INSERT INTO table
    def insert(self, table, values):
        if isinstance(values, tuple):
            values = [values];
        if not isinstance(values, list):
            raise TypeError("INSERT: Inappropriate argument type for parameter values");

        cols = map(lambda x: self.enclose_sys(x[0]), values);
        vals = tuple(map(lambda x: x[1], values));

        sql = 'INSERT INTO %s(%s) VALUES (%s)' % (self.enclose_sys(table), ','.join(cols), ','.join( ['%s'] * len(vals) ));
        return self._query(sql, vals);

    #UPDATE table
    def update(self, table, values, where = None, op = 'AND'):
        if isinstance(values, tuple):
            values = [values];
        if not isinstance(values, list):
            raise TypeError("UPDATE: Inappropriate argument type for parameter values");

        cols = map(lambda x: self.enclose_sys(x[0])+'=%s', values);
        vals = tuple(map(lambda x: x[1], values));

        (where_clause, where_params) = self.__expand_where_clause(where, op);

        if where_clause:
            return self._query('UPDATE %s SET %s WHERE %s' % (self.enclose_sys(table), ','.join(cols), where_clause), list(vals) + list(where_params));
        else:
            return self._query('UPDATE %s SET %s' % (self.enclose_sys(table), ','.join(cols)), vals);

    #DELETE FROM table
    def delete(self, table, where = None, op = 'AND'):
        (where_clause, where_params) = self.__expand_where_clause(where, op);

        if where_clause:
            return self._query("DELETE FROM %s WHERE %s" % (self.enclose_sys(table), where_clause), where_params);
        else:
            return self._query("DELETE FROM %s" % (self.enclose_sys(table)));

    #upsert and merge perform the same task, having the same end result.
    #The difference is that the former is optimised to work on data where usually little new rows are added
    #while the latter is optimised in the case the majority of the data dealt with will be added, not already existing
    def upsert(self, table, values, where):
        self.update(table, values, where);
        if not self.affected_rows():
            self.insert(table, [values] + [where]);
    def merge(self, table, values, where):
        try:
            self.insert(table, [values] + [where]);
        except self._module.DatabaseError as e:
            #TODO: Check error in case it's not due to a PK/Unique violation
            self.update(table, values, where);

    #Returns a row, instead of simply a list of 1. Inspired by Wordpress
    def get_row(self, table, columns = None, where = None, op = "AND"):
        r = self.select(table, columns, where, op);
        return r[0] if r else None;

    #Returns a variable. Useful for quick counts or returning of an id, for example. Inspired by Wordpress
    def get_var(self, table, columns = None, where = None, op = "AND"):
        r = self.select(table, columns, where, op);
        return r[0].items()[0][1] if r else None;

    #Count the rows of a table
    def count(self, table, column, value = None):
        where = (column, value) if value else None;
        return self.get_var(table, 'COUNT(*) AS %s' % (self.enclose_sys('cunt')), where);


    def drop(self):
        self._query("DROP DATABASE " + self._dbname);
    def create(self):
        self._query("CREATE DATABASE " + self._dbname);
    def purge(self):
        #only works in MySQL, must find alternative for Postgres
        self.drop();
        self.create();
    def truncate(self, table_name):
        self._query("TRUNCATE TABLE " + self.enclose_sys(table_name));


    #wrappers around transaction management functions
    def commit(self):
        self._con.commit();
    def rollback(self):
        self._con.rollback();
    def autocommit(self, val):
        self._autocommit(bool(val));

    #getters... 
    def affected_rows(self):
        return self._affected_rows; 
    def insert_id(self):
        return self._insert_id;

    def __is_escaped(self, s, pos):
        for char in ["'", "\\"]:
            j = pos - 1;
            count = 0;

            #count back the num. of appearances of certain char
            while (j>=0 and s[j] == char):
                j-=1;
                count+=1;

            #reduce the count in cases like \'' ,where the last ' to the left is escaped by 1 or more \
            if (char == "'" and count and self.__isEscaped(s, pos-count)):
                count-=1;
            if (count):
                break;

        return True if (count % 2) else False;


    #replaces MySQL style `backticks` with "double quotes", as per SQL standard.
    #Required in order to support MySQL queries containing backticks
    def __replace_backticks(self, str):
        s = list(str);
        delim = None;
        inside = False;


        for i in range(0, len(s)):
            #only working on important characters
            if (s[i] not in ['"',"'","`"]):
                continue;

            if inside:
                if (s[i] == '`' or s[i] != delim): #if we encounter a wrong token, simply continue
                    continue;

                if not self.__is_escaped(s, i):
                    inside = False;
                    delim = None;
            else:
                if s[i] == '`':
                    s[i] = '"';
                    continue;

                if not self.__is_escaped(s, i):
                    inside = True;
                    delim = s[i];

        return "".join(s);

    #helper function, expands a tuple/list of tuples containing where parameters to string
    def __expand_where_clause(self, where, op):
        params = [];
        clauses = [];

        if where:
            if isinstance(where, tuple):
                where = [where];
            if not isinstance(where, list):
                raise TypeException("Unknown type for WHERE clause argument");

        if where:   
            for clause in where:
                clause_op = clause[2] if len(clause)==3 else '=';
                clauses.append(self.enclose_sys(clause[0]) + (" %s " % clause_op) + '%s');
                params.append(clause[1]);

        where_clause = (' %s ' % op).join(clauses);
        return (where_clause, tuple(params) if len(params) else None);  

    #returns an array containing the names of the columns of a table
    def table_columns(self, table_name = None, cur = None):
        if not cur:
            try:
                cur = self._con.cursor();
                cur.execute("SELECT * FROM " + table_name + " LIMIT 1");
                cur.close();
            except self._module.DatabaseError as e:
                raise Exception("Database Error: %s" % str(e));

        cols = map(lambda x: x[0], cur.description);
        return cols;

Answer 1

1. Comments like # Will be removed soon make me think that this isn't really ready for review. It's best to fix all the problems you know about before submitting for review.

2. There's no documentation. How is anyone supposed to use this? For example, what arguments should one pass to the constructor? What kind of object do I pass in if I want a where clause?

3. Why inherit from Singleton? Does this mean I can't connect to more than one database? Isn't that a problem?

4. There are SQL injection vulnerabilities everywhere, because enclose_sys() doesn't escape its argument if it contains a space. For example, if I run:

    db = Database(username='foo', password='bar', database='mydb', driver='mysql')
    db.select('table', ('1; drop database; select *',))
   

   then this issues the query

    SELECT 1; drop database; select * FROM `table`
   

5. There's no separation of driver code into its own class. If I want to add a new backend (e.g. SQLite) how do I do it?

6. Your upsert() method doesn't take advantage of the capability of some databases to do it in a single query, for example INSERT ... ON DUPLICATE KEY UPDATE in MySQL.

7. In disconnect() you don't clear self._con, so calling disconnect() twice will cause you to close the connection twice.

8. It's better to write values = list(values) instead of testing isinstance(values, list) and then raising an exception if it's not. The former can accept iterators and generators.

9. In select(), why does the columns argument have to be a tuple? Why not a list?

10. Basically if isinstance(arg, type1): ... elif isinstance(arg, type2): ... is an anti-pattern in Python. It usually means that your interface is poorly thought out and you are trying to overload the meaning of your parameter. Better to choose a single type for each parameter.

11. You raise TypeException in various places, but there is no such exception in Python. Presumably this is a typo for TypeError, but it suggests that you have never tested these error paths.

12. Your _query() method always returns its results as a list of dictionaries, thus reading the entire result set into memory. This defeats the whole point of having database cursors, which is that the application can process records one at a time as they are returned from the database. This allows the application to proceed in parallel with the database, and avoids filling memory unnecessarily.

13. It's not very portable to Python 3: kwargs.has_key('host'), basestring.

14. You use Python's private variables in a couple of places: __is_escaped, __expand_where_clause, __replace_backticks. This makes it impossible to test or experiment with these methods. You should use private variables only when you absolutely have to to avoid name clashes.

15. There's a method enclose_str but it does not seem to be called.

16. Many more problems which I have not the will to type up ... basically this is far from ready for anyone to use.

Responding to your comments:

• To #3, you say, "I needed a way to pass a 'main' database connection to other modules". But this doesn't explain why Database inherits from Singleton. Why can't you just write:

  main_db = Database(...)
  

  and then pass main_db to the other modules? You should beware of letting a local concern (your program needs a main database connection) influence the design of a module that is supposed to be general.

  You go on to say, "you can have how many you instances you like". But in the singleton pattern there can be at most one instance of the class. Hence the name. So I don't understand this remark.

• To #4, you indicate that you don't think it's the responsibility of your class to avoid SQL injection attacks. That's fair enough, but one of the most important reasons to use an object-relational mapping system is that it takes care of this task, which is difficult and tedious to get right.

• To #10, you say, "I'd rather have it take multiple types and return multiple types than write 3 separate functions." Yes, of course, as an implementor you want to make things easy for yourself. But I'm looking at your code from the perspective of a user or maintainer, and I would prefer interfaces to be simple and easy to understand.

  In any case, how hard is it to write:

  def escape_list(self, seq):
      """Escape all the elements in seq, and return them as a list."""
      return map(self.escape, seq)
  

Answer 2

There's quite a lot of code so I'm going to comment only this small piece of code :

results = cur.fetchall();
n = len(results);
cols = self.table_columns(None, cur);
retval = [];
for i in range(0,n):
    aux = results[i];
    row = {};
    for j in range(0,len(cols)):
        row[cols[j]] = aux[j];
    if len(row):
        retval.append(row);
return retval

1. You don't need ; at the end of the lines
2. The best way to iterate on something in Python is via for element in something (or for element,index in enumerate(something) if you need the index).
3. You can use the fact that the boolean evaluation of an empty container is false to write if row instead of if len(row).

Your code is now :

results = cur.fetchall();
cols = self.table_columns(None, cur);
retval = [];
for res in results:
    row = {};
    for col,j in enumerate(cols):
        row[col] = res[j];
    if row:
        retval.append(row);
return retval

Now, you can create your row dictionnary direclty from res and cols (other solutions can be found on SO):

results = cur.fetchall();
cols = self.table_columns(None, cur);
retval = [];
for res in results:
    row = dict(zip(cols,res))
    if row:
        retval.append(row);
return retval

Now, there would be a way to make the whole thing a single expression using list comprehension but I am not quite sure this is a good idea.