Pg : add some SQL helpers

This commit is contained in:
Benjamin Renard 2018-12-12 17:17:56 +01:00
parent 93d2133376
commit 41a2dde02c

117
PgDB.py
View file

@ -1,8 +1,11 @@
#!/usr/bin/python #!/usr/bin/python
# -*- coding: utf-8 -*-
import psycopg2 import psycopg2
import logging import logging
import sys import sys
import traceback
import datetime
class PgDB(object): class PgDB(object):
@ -13,6 +16,9 @@ class PgDB(object):
con = 0 con = 0
date_format = '%Y-%m-%d'
datetime_format = '%Y-%m-%d %H:%M:%S'
def __init__(self,host,user,pwd,db): def __init__(self,host,user,pwd,db):
self.host = host self.host = host
self.user = user self.user = user
@ -51,7 +57,7 @@ class PgDB(object):
self.con.commit() self.con.commit()
return True return True
except Exception, e: except Exception, e:
logging.error('Erreur durant la requete sql %s : %s' % (sql,e)) logging.error(u'Error during SQL request "%s" :\n\n%s' % (sql.decode('utf-8', 'ignore'), traceback.format_exc()))
self.con.rollback() self.con.rollback()
return False return False
@ -62,5 +68,112 @@ class PgDB(object):
results = cursor.fetchall() results = cursor.fetchall()
return results return results
except Exception, e: except Exception, e:
logging.error('Erreur durant la requete sql %s : %s' % (sql,e)) logging.error(u'Error during SQL request "%s" :\n\n%s' % (sql.decode('utf-8', 'ignore'), traceback.format_exc()))
return False return False
#
# SQL helpers
#
def _quote_value(self, value):
if type(value) is int or type(value) is float:
return unicode(value)
if type(value) is str:
value = unicode(value)
elif type(value) is datetime.datetime:
value = unicode(self._format_datetime(value))
elif type(value) is datetime.date:
value = unicode(self._format_date(value))
return u"'%s'" % value.replace(u"'",u"''")
def _format_where_clauses(self, where_clauses, where_op=u'AND'):
if type(where_clauses) is str:
return where_clauses
elif type(where_clauses) is list:
return (u" %s " % where_op).join(where_clauses)
elif type(where_clauses) is dict:
return (u" %s " % where_op).join(map(lambda x: "%s=%s" % (x, _quote_value(where_clauses[x])), where_clauses))
logging.error('Unsupported where clauses type %s' % type(where_clauses))
return False
def _format_datetime(self, datetime):
return datetime.strftime(self.datetime_format)
def _format_date(self, date):
return date.strftime(self.date_format)
def time2datetime(self, time):
return self._format_datetime(datetime.fromtimestamp(int(time)))
def time2date(self, time):
return self._format_date(datetime.fromtimestamp(int(time)))
def insert(self, table, values, just_try=False):
sql=u"INSERT INTO %s (%s) VALUES (%s)" % (table, u', '.join(values.keys()), u", ".join(map(lambda x: self._quote_value(values[x]), values)))
if just_try:
logging.debug(u"Just-try mode : execute INSERT query : %s" % sql)
return True
logging.debug(sql)
if not self.doSQL(sql):
logging.error(u"Fail to execute INSERT query (SQL : %s)" % sql)
return False
return True
def update(self, table, values, where_clauses, where_op=u'AND', just_try=False):
where=self._format_where_clauses(where_clauses, where_op=where_op)
if not where:
return False
sql=u"UPDATE %s SET %s WHERE %s" % (table, u", ".join(map(lambda x: "%s=%s" % (x, self._quote_value(values[x])), values)), where)
if just_try:
logging.debug(u"Just-try mode : execute UPDATE query : %s" % sql)
return True
logging.debug(sql)
if not self.doSQL(sql):
logging.error(u"Fail to execute UPDATE query (SQL : %s)" % sql)
return False
return True
def delete(self, table, where_clauses, where_op=u'AND', just_try=False):
where=self._format_where_clauses(where_clauses, where_op=where_op)
if not where:
return False
sql=u"DELETE FROM %s WHERE %s" % (table, where)
if just_try:
logging.debug(u"Just-try mode : execute DELETE query : %s" % sql)
return True
logging.debug(sql)
if not self.doSQL(sql):
logging.error(u"Fail to execute DELETE query (SQL : %s)" % sql)
return False
return True
def select(self, table, where_clauses=None, fields=None, where_op=u'AND', order_by=None, just_try=False):
sql = u"SELECT "
if fields is None:
sql += "*"
elif type(fields) in (str, unicode):
sql += fields
else:
sql += u", ".join(fields)
sql += u" FROM " + table
if where_clauses:
where=self._format_where_clauses(where_clauses, where_op=where_op)
if not where:
return False
sql += u" WHERE " + where
if order_by:
sql += u"ORDER %s" % order_by
return self.doSelect(sql)