# Database abstraction class

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._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['db'] = self._dbname;
elif self._driver == 'pgsql':
kwargs['database'] = self._dbname;

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;

• Glancing through, I don't quite see your rationale of using **kwargs** If _username and others are mandatory fields, why not just have the argument explicitly passed through the constructor? – nz_21 Jan 21 at 16:48

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.

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

• First, thanks for the answer. Since I'm a bit of a newbie, this helps a lot. All of your points are valid, though I'd like to discuss a few 3.I needed a way to pass a "main" database connection to other modules. Appart from that, you can have how many you instances you like. 4. enclose_sys doesn't escape any parameter. If one needs the marginal case of allowing users access to the structure of the table, he should do the escaping himself(should be written in the docs though). 10.E.g."escape", I'd rather have it take multiple types and return multiple types than write 3 separate functions. – Vlad Apr 9 '13 at 1:31
• 3.Singleton is an bad name choice, I do agree, but I couldn't find a better one. It's more of a "main/default instance, but allow instantiation through the constructor as well if you need another". The whole Singleton class itself is just a bunch of static methods anyway(get/set/clear). 4.Yes it is, but I've yet to hear of escaping system identifiers(column names). Solved it by removing whitespace check & raising when I encounter a " character. 10.I'm not lazy, I just thought it was a more elegant solution. E.g. the list function, which does 2 things(unpacks if iterable, list of 1 if not). – Vlad Apr 9 '13 at 10:35

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

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.