SQL Helper Functions

[bartonc]

Here is the first installment of SQL helper functions which I use with two classes that I wrote which encapsulate a MySQL server and a MySQL client. These helpers make it easy to convert back and forth between python dictionaries and SQL tables.

def MySQLSelect(table, arglist=[], argdict={}, **kwargs):
    """Build an SQL SELECT command from the arguments:
    Return a single string which can be 'execute'd.
    arglist is a list of strings that are column names to get.
    argdict and kwargs are two way to evaluate 'colName'=value
    for the WHERE clause"""
    a = ', '.join(arglist)
    args = argdict.copy()
    args.update(kwargs)
    for key, value in args.items():
        args[key] = (str(value), repr(value))[type(value) == str]
    b = ''
    if args:
        b = 'WHERE %s' % ' AND '.join(key + '=' + value
                                     for key, value in args.items())
    return ' '.join(['SELECT', (a or '*'), 'FROM', table, b])
 

[bartonc]

Here is the next installment:

 
def MySQLInsert(table, argdict={}, **kwargs):
    """Build an SQL INSERT command from the arguments:
    Return a single string which can be 'execute'd.
    argdict is a dictionary of 'column_name':value items.
    **kwargs is the same but passed in as column_name=value"""
    args = argdict.copy()
    args.update(kwargs)
    keys = args.keys()
    argslist = []
    for key in keys:
        a = args[key]
        argslist.append((str(a), repr(a))[type(a) == str])
    a = '(%s)' %', '.join(field for field in keys)
    b = '(%s)' %', '.join(argslist)
    return ' '.join(['INSERT', table, a, 'VALUES', b])
 

[bartonc]

Here's the update converter:

 
 
def MySQLUpdate(table, valuedict, argdict={}, **kwargs):
    """Build an SQL SELECT command from the arguments:
    Return a single string which can be 'execute'd.
    valuedict is a dictionary of column_names:value to update.
    argdict and kwargs are two way to evaluate 'colName'=value
    for the WHERE clause."""
    vargs = valuedict.copy()
    for key, value in vargs.items():
        vargs[key] = (str(value), repr(value))[type(value) == str]
    a = 'SET %s' % ', '.join(key + '=' + value
                                 for key, value in vargs.items())
    args = argdict.copy()
    args.update(kwargs)
    for key, value in args.items():
        args[key] = (str(value), repr(value))[type(value) == str]
    b = ''
    if args:
        b = 'WHERE %s' % ' AND '.join(key + '=' + value
                                     for key, value in args.items())
 
    return ' '.join(['UPDATE', table, a, b])

[bartonc]

Here's a pretty simplistic delete fuction;

 
 
def MySQLDelete(table, argdict={}, **kwargs):
    """Build an SQL DELETE command from the arguments:
    Return a single string which can be 'execute'd.
    argdict and kwargs are two way to evaluate 'colName':value
    for the WHERE clause."""
    args = argdict.copy()
    args.update(kwargs)
    for key, value in args.items():
        args[key] = (str(value), repr(value))[type(value) == str]
    b = ''
    if args:
        b = 'WHERE %s' % ' AND '.join(key + '=' + value
                                     for key, value in args.items())
    return ' '.join(['DELETE FROM', table, b])
 
 

[bartonc]

And here is a class that I call a server because it stores the connection and cursor, handles errors and sends info (last query, etc) to its master's write() command. called with sys.stdout as the master, my dbServer prints to stdout.

from MySQLdb import *
from time import time
 
class DBServer:
    def __init__(self, master):
        self.master = master
 
    def Login(self, servername, username, password, database=""):
        try:
            self.dbconnect = connect(host=servername, user=username, passwd=password, db=database)
        except (DatabaseError, OperationalError):
            self.dbconnect = None
            self.master.write('Couldn\'t connect to the database named %s on host %s'
                              %(repr(database), repr(servername)))
            return
        self.dbcursor = self.dbconnect.cursor()
        self.Execute('SET autocommit=1')
        return self.dbconnect
 
    def DBError(self):
        """Remove the current message from the cursor
           and display it."""
        try:
            (error, message) = self.dbcursor.messages.pop()
        except AttributeError:
            (error, message) = self.dbconnect.messages.pop()
        self.master.write('%s #%d:  %s' %(str(error).split('.')[-1],
                                          message[0], message[1]))
 
    def Execute(self, query):
        try:
            cursor = self.dbcursor
            now = time()
            cursor.execute(query)
            nRows = cursor.rowcount
            self.master.write(query)
            self.master.write("%d rows affected: %.2f sec." %(nRows, time() - now))
        except (DatabaseError, OperationalError):
            self.DBError()
            return
        return cursor
 
    def DBExists(self, database):
        """Return True if database exists"""
        resultset = self.Execute("show databases").fetchall()
        ## print resultset
        return (database,) in resultset
 
    def close(self):
        self.dbconnect.close()
 

[bartonc]

This topic is currently being discussed, so I'm bumping this thread.

[bartonc]

This topic is currently being discussed, so I'm bumping this thread.

These have recently been updated in this thread in the Articles section.