python连接SQL SERVER数据库：
Python编程中可以使用SQL SERVER 进行数据库的连接及诸如查询/插入/更新等操作，但是每次连接SQL SERVER 数据库请求时，都是独立的去请求访问，相当浪费资源，而且访问数量达到一定数量时，对SQL SERVER 的性能会产生较大的影响。因此，实际使用中，通常会使用数据库的连接池技术，来访问数据库达到资源复用的目的。
python的数据库连接池包 DBUtils：
DBUtils是一套Python数据库连接池包，并允许对非线程安全的数据库接口进行线程安全包装。DBUtils来自Webware for Python。
DBUtils提供两种外部接口：
- * PersistentDB ：提供线程专用的数据库连接，并自动管理连接。
- * PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。
需要的python库：
下载DBUtils：
Webware 的网站下载最新版本：http://www.webwareforpython.org/downloads/DBUtils/
或者在Python Package Index来下载：http://www.python.org/pypi/DBUtils/
下载pymssql：
http://code.google.com/p/pymssql/downloads/list
(pymssql 是Python语言用来连接微软 SQL SERVER 数据库的类库)
1.写一个创建连接池，获取连接以及重新连接数据库的模块：
# libby_db_pool.py
# 代码如下：
#-*- coding:utf-8 -*-
from DBUtils.PooledDB import PooledDB
import pymssql #sqlserver数据库适配器 from pymssql import OperationalError, InternalError, ProgrammingError HOST = "127.0.0.1"
PORT = "1433"
CHARSET = "utf8"
NAME = "zkeco_oracle"
USER = "sa"
PASSWORK = "sa" conn_args = {
'host':"%s"%HOST,
'port':"%s"%PORT,
'database':"%s"%NAME,
'charset':"%s"%CHARSET,
'user':"%s"%USER,
'password':"%s"%PASSWORK
} """
mincached : 启动时开启的闲置连接数量(默认值 0，意味着开始时不创建数据库连接)
maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小，对于线程安全的数据库，这个不起作用)
maxshared : 共享连接数允许的最大数量(默认值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用
maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)
blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......>; 其他代表阻塞直到连接数减少,连接被分配)
maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)
setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...] """ args = (10,10,30,100,True,0,None) class DbManager():
def __init__(self):
try:
self._pool = PooledDB(pymssql,*args,**conn_args)
except Exception,e:
print "The parameters for DBUtils is:",conn_args
def _getConn(self):
return self._pool.connection() _dbManager = DbManager() def getConn():
""" 获取数据库连接 """
return _dbManager._getConn() def _reConn():
""" 重新连接数据库 """
global _dbManager
re = False
try:
_dbManager = DbManager()
re = True
except:
import traceback
traceback.print_exc()
finally:
return re
import datetime def reConn():
print "%s: now try to reconnect Database!"%(datatime.datatime.now())
flag = _reConn()
if flag:
print "%s reconnect database success!"%(datatime.datatime.now())
else:
print "%s reconnect database failed!"%(datatime.datatime.now())
2.写一个支持增删查改功能的连接池模块：
#libby_sql_utils.py
#代码如下： #-*- coding:utf-8 -*-
from libby_db_pool import getConn,reConn
from libby_db_pool import OperationalError, InternalError, ProgrammingError
import traceback
def test_conn():
"""
测试连接池连接是否正常
return：
res：True：正常，False：不正常
msg：如果不正常，为异常信息
""" test_sql = """
select 1 """ conn = None
cur = None
res = False
msg = ""
try:
conn = getConn()
cur = conn.cursor()
cur.execute(test_sql)
res = cur.fetchall()
res = True
except Exception,e:
trackback.print_exc()
msg = e
finally:
if cur:
cur.close()
if conn:
conn.close()
return res,msg def call_reConn():
"""
重新创建连接池
"""
reConn() def p_query(sql):
"""
dbutils 数据连接池
只能执行数据查询sql语句,否则会抛错
@parm: 要执行的sql语句
@return:
[]:查询结果为空
None:sql语句执行失败,出现异常
二维list:正常结果
""" conn = None
cur = None
res = None
try:
conn = getConn()
cur = conn.cursor()
cur.execute(sql)
res = cur.fetchall()
except (OperationalError, InternalError):
call_reConn()
trackback.print_exc()
except:
trackback.exc()
finally:
if cur:
cur.close()
if conn:
conn.close()
return res def p_query_one(sql):
"""
dbutils 数据连接池
只能执行数据查询sql语句，否则会报错
执行sql查询语句，获取第一条记录
@parm：要执行的sql语句
@return：
[]:查询结果为空
None:sql语句执行失败,出现异常
list:正常结果
""" conn = None
cur = None
res = None
try:
conn = getConn()
cur = conn.cursor()
cur.execute(sql)
res = cur.fetchone()
except (OperationalError,InternalError):
call_reConn()
except:
traceback.print_exc()
finally:
if cur:
cur.close()
if conn:
conn.close()
return res def p_execute(sql):
"""
dbutils 数据连接池
执行数据操作语句，包括 update,insert,delete
@parm:要执行的sql
@return:
None:sql语句执行失败,出现异常
number:影响记录条数
-2:数据库连接失败导致执行失败
"""
conn = None
cur = None
res = None
try:
conn = getConn()
cur = conn.cursor()
cur.execute(sql)
res = cur._cursor.rowcount
conn.commit()
except Exception,e:
if conn:
conn.rollback()
traceback.print_exc()
finally:
if res == -1:#可能是数据库断开连接
ret,msg = test_conn()
if not ret:
call_reConn()
res = -2
if cur:
cur.close()
if conn:
conn.close()
return res def p_mutiexec(sql_list):
"""
dbutils 数据连接池
执行多条数据操作语句,可以用于多条sql语句的事务性操作,包括 update,insert,delete @parm:要执行的sql语句[]
@return:
(flag,res):
flag<Ture or False>:批次是否全部执行成功
res<list>:每天sql语句执行影响的行数,如果执行失败,由此可以判断第几条sql语句执行失败
如果遇到数据库断开的情况,返回[-2,] """ conn = None
cur = None
res = []
flag = True try:
conn = getConn()
cur = conn.cursor()
for sql in sql_list:
cur.execute(sql)
num = cur._cursor.rowcount
res.append(num)
conn.commit()
except Exception,e:
if conn:
conn.rollback()
traceback.print_exc()
finally:
if -1 in res:
ret,msg = test_conn()
if not ret:
call_reConn()
flag = False
res = [-2,]
if cur:
cur.close()
if conn:
conn.close()
return flag,res