6,596
There are some issues to resolve in supporting multiple connection types and I plan to add PySQLite to the mix. The this update is due to my suggesting the accompanying helper functions in a recent post and noticing that the ones posted here are a bit out of date. I'll post those helpers in a separate thread.
I've also been working with the following database engines:
MySQL 5.x
SQLite3
JET engine 4.0
Here is the latest dbConection container which was originally called dbServer, but I'd love suggestions for a better name:
Expand|Select|Wrap|Line Numbers
- #-----------------------------------------------------------------------------
- # Name: SQLConMan.py ### This needs a better name ###
- # Purpose: database connection handling. The DBServer() traps all errors and
- # writes status to its master (which can be stdout). If an error does
- # occure, Execute() returns None. This might be confusing since pyodbc
- # returns None from execute() on quries that don't return a result set.
- # It's interesting to not that db interface creator have begun add an
- # execute() funtion to their connection objects.
- #
- # Author: <your name>
- #
- # Created: 2007/10/13
- # RCS-ID: $Id: wxdbtools.py $
- # Copyright: (c) 2007
- # Licence: <your licence>
- #-----------------------------------------------------------------------------
- from time import time
- ## MySQL is the default db here:
- ## MySQL differs from MSSQL and SQLite regarding TRANSACTION statemets ##
- TRANSACTIONS = ('START TRANSACTION', 'BEGIN TRANSACTION') # MySQL, SQLite & MSSQL
- TransactionQuery = TRANSACTIONS[0]
- ## and these have a different matrix of support
- LASTINSERTQUERIES = ('SELECT Last_Insert_Id()', 'SELECT @@IDENTITY') # MySQL & SQLite, MSSQL
- LastInsertQuery = LASTINSERTQUERIES[0]
- ## One idea for detecting the DB type is to try each one until it works ##
- PYODBC = 'pyodbc'
- MXODBC = 'mxodbc'
- WIN23ODBC = 'win32odbc'
- class SQLConMan(object):
- global TransactionQuery
- def __init__(self, master):
- self.master = master # a place to call write() on
- self.dbConnection = None # the API 2.0 connection object
- self.dbCursor = None # the API 2.0 cursor object
- def __getattribute__(self, name):
- """Redirect method calls to the connection 'object'."""
- try:
- return object.__getattribute__(self, name)
- except AttributeError:
- # __getattribute__() only work for subtypes of object.
- ## return object.__getattribute__(self.dbConnection, name)
- return eval("self.dbConnection.%s" %(name))
- def Login(self, servername, username, password, autocommit=1, database="", conType=PYODBC):
- """Attempt to create a database login. If successful, return
- an open connection. Otherwise, return None."""
- global OperationalError, ProgrammingError, DatabaseError, DriverError
- odbcstring = "DSN=%s;UID=%s;PWD=%s" %(servername, username, password)
- if conType == PYODBC:
- from pyodbc import connect
- from pyodbc import OperationalError
- from pyodbc import ProgrammingError
- from pyodbc import DatabaseError
- from pyodbc import Error as DriverError
- cac = int(not autocommit)
- try:
- self.dbConnection = connect(odbcstring, autocommit=autocommit)
- self.dbCursor = self.dbConnection.cursor()
- except (DatabaseError, OperationalError, DriverError), message:
- self.NoLogin(servername, username, message)
- return
- elif conType == MXODBC:
- from mx.ODBC.Windows import Connect
- from mx.ODBC.Windows import OperationalError
- from mx.ODBC.Windows import ProgrammingError
- from mx.ODBC.Windows import DatabaseError
- from mx.ODBC.Windows import error as DriverError
- cac = int(not autocommit)
- try:
- self.dbConnection = Connect(servername, user=username, password=password,
- clear_auto_commit=cac)
- self.dbCursor = self.dbConnection.cursor()
- except (DatabaseError, OperationalError), message:
- self.NoLogin(servername, username, message)
- return
- elif conType == WIN32ODBC:
- import odbc
- from dbi import opError as OperationalError
- from dbi import progError as ProgrammingError
- from dbi import dataError as DatabaseError
- from odbc import OdbcError as DriverError
- try:
- self.dbConnection = odbc.odbc(odbcstring)
- self.dbCursor = self.dbConnection.cursor()
- except (DatabaseError, OperationalError), message:
- self.NoLogin(servername, username, message)
- return
- self.master.write("%s has been logged onto %s\n" %(username, servername))
- if database:
- try:
- self.dbCursor('USE %s' %database)
- except (DatabaseError, OperationalError):
- pass
- return self.dbConnection
- def NoLogin(self, servername, username, message):
- self.master.write('Couldn\'t log on to the server `%s` as `%s`\n' %(servername, username))
- self.DBError("", message)
- def DBError(self, query, message):
- """Format the current message and display it.
- Report the query and error to the master."""
- self.master.write('ODBC Error: %s\n' %message)
- ## raise
- # all raised by PyODBC from various DBs #
- # """('42000', "[42000] [Microsoft][ODBC Microsoft Access Driver]
- # Invalid SQL statement; expected 'DELETE', 'INSERT', 'PROCEDURE', 'SELECT', or 'UPDATE'. (-3500)")"""
- # """ ('42S02', "[42S02] [Microsoft][ODBC Microsoft Access Driver]
- # The Microsoft Jet database engine cannot find the input table or query 'main'.
- # Make sure it exists and that its name is spelled correctly. (-1305)")"""
- # """('42S02', "[42S02] [MySQL][ODBC 3.51 Driver][mysqld-5.0.22-community-max-nt]
- # Table 'hetap_setup.main' doesn't exist (1146)")"""
- # """('HY000', '[HY000] near "USE": syntax error (1) (1)')"""
- # """('HY000', '[HY000] no such table: main (1) (1)')"""
- ## self.master.write('%s\nODBC Error %s #%d: %s #%d\n' %(query, message[0], message[1],
- ## message[-2].split(']')[-1], message[-1]))
- def Execute(self, query, *params):
- """Execution method reports on the number of rows affected and duration
- of the database query execution and catches errors. Return a reference
- to the cursor if no error ocurred, otherwise, None."""
- cursor = self.dbCursor
- if cursor:
- try:
- now = time()
- if params:
- cursor.execute(query, *params)
- else:
- cursor.execute(query)
- ## print cursor
- ## nRows = cursor.rowcount
- ## self.master.write("%s " % query)
- ## self.master.write("%d rows affected: %.2f sec.\n" %(nRows, time() - now))
- except (DatabaseError, OperationalError, DriverError), message:
- self.DBError(query, message)
- return
- return cursor
- def ExecuteScript(self, query):
- """"""
- for singlequery in query.split(';'):
- if singlequery:
- self.Execute(singlequery + ';')
- def DBExists(self, database):
- """Return True if database exists"""
- cursor = self.Execute("show databases")
- if cursor:
- rows = cursor.fetchall()
- return (database.strip('`').lower(),) in rows
- def TableExists(self, table):
- """Return True if database exists"""
- cursor = self.Execute("show tables")
- if cursor:
- rows = cursor.fetchall()
- return (table.strip('`').lower(),) in rows
- def ConnectionIsOpen(self):
- ## """Return 1 if the connection has been established and is not closed.
- ## Return 0 if it is closed; None, if it has not been established."""
- return self.dbConnection is not None
- def CloseConnection(self):
- """Close the connection if it exists and is open."""
- if self.dbConnection is not None:
- try:
- self.dbConnection.close()
- self.master.write("Closed connection.\n")
- except ProgrammingError:
- self.master.write("Connection is already closed!\n")
- else:
- self.master.write("Connection does not exist!\n")
- def SetMaster(self, master):
- """Allow the master to be reset."""
- self.master = master
- ## May want to rethink these ones ##
- def GetMaster(self):
- return self.master
- def GetDbConnection(self):
- return self.dbConnection
