DB连接管理器现在支持mxODBC，win32ODBC和PyODBC
茹轩昂
这是一项正在进行的工作（当前和进行中）。
支持多种连接类型需要解决一些问题，我计划将PySQLite添加到其中。 此更新是由于我在最近的帖子中建议了附带的帮助程序功能，并注意到此处发布的助手功能有些过时了。 我将这些帮助程序发布在单独的线程中。
我还一直在使用以下数据库引擎：
MySQL 5.x
SQLite3
JET引擎4.0
这是最初称为dbServer的最新dbConection容器，但是我很乐意建议使用更好的名称：
#-----------------------------------------------------------------------------
# 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
类似资料：