Python使⽤DBUtils连接部分主流数据库
需要了解的知识
⾸先呢，你需要了解下，我的描述肯定没官⽹解释来的清晰，⾃⾏阅读后，你就会发现我为什么会选⽤PooledDB⽽不是其他作为连接池了。
其次，DBUtils⽀持所有遵循规范的数据库连接模块，也就是说除了我⽰例中所提供的⼏个数据库连接⽅式外，各位可以探索其他遵循此标准的连接模块，从⽽在此基础上拓展，成为连接更多种类数据库的通⽤⼯具类。
最后，以下内容均基于python3。
---------------------------------------------------------
10.23增补内容：⽀持hba，更新字典返回⽅式以及部分⽅法扩展。
准备⼯作
⾸先，pip下⽀持DB-API 2规范的相关数据库连接模块和DBUtils。
pip install DBUtils
pip install pymysql（mysql）
pip install pymssql（sqlrver）
pip install cx_Oracle（oracle）
pip install phoenixdb（hba）
pip install sqlite3（sqlite3 python⾃带）
其次，需要准备⼀份配置⽂件，姑且命名为pdbc.properties，以下是⽰例，根据数据库的连接信息进⾏修改即可。
# 数据库类型，⽀持mysql,oracle,sqlrver,sqlite3,hba
# --------------------------------------------------------------
# mysql
# 连接数据库host
host_mysql=ip
# 连接数据库port
port_mysql=3306
# 连接数据库库名
databa_mysql=dbname
# ⽤户
ur_mysql=urname
# 密码
password_mysql=password
# 字符集
chart_mysql=utf8
# --------------------------------------------------------------
# oracle
# 连接数据库host
host_orc=ip
# 连接数据库port
port_orc=1521
# 连接数据库库名
databa_orc=dbname
# ⽤户
ur_orc=urname
# 密码
password_orc=password
# 字符集
nencoding_orc=utf8
# --------------------------------------------------------------
# sqlrver
# 连接数据库host
host_ms=ip
# 连接数据库port
port_ms=1433
# 连接数据库库名
databa_ms=dbname
# ⽤户
ur_ms=urname
# 密码
password_ms=password
# 字符集
chart_ms=utf8
# --------------------------------------------------------------
# sqlite3
# 连接数据库⽂件名，sqlite不⽀持加密，不使⽤⽤户名和密码
databa_sqlite3=path/to/your/dbname.db
# --------------------------------------------------------------
# hba
# 连接数据库host
host_hb=ip
# 连接数据库port，phoenixdb连接使⽤8765端⼝⽽⾮2181等其他端⼝
port_hb=8765
# ⽤户
ur_hb=urname
# 密码
password_hb=password
然后，准备⼀份读取properties⽂件的⼯具类，姑且称为PropertiesUtil.py，可以寻找⽹上的，也可以参考我写的。
# -*- coding:utf-8 -*-
class PropertiesUtil(object):
# 缓存配置
__file_dict = {}
def get_config_dict(lf, file_path="pdbc.properties"):
"""
获取资源⽂件，形成字典
:param file_path: ⽂件路径
:return：字典内容的key、value均为字符串
"""
if file_path not in lf.__file_dict:
properties = {}
with open(file_path, 'r', encoding='UTF-8') as pro_file:
for line in adlines():
line = line.strip().replace('\n', '')
if line.find('=') > 0 and not line.startswith('#'):
strs = line.split('=')
value = line[len(strs[0]) + 1:]
lf.__get_dict(strs[0].strip(), properties, value.strip()) lf.__file_dict[file_path] = properties
return lf.__file_dict[file_path]
def get_config_value(lf, file_path, prop_name):
"""
获取资源⽂件，形成字典，获取属性值
:
param file_path: ⽂件路径
:param prop_name: 属性名称
:return: 返回字符串格式的属性值
"""
_config_dict(file_path)[prop_name]
def __get_dict(lf, dict_name, properties, value):
"""
递归获取配置字典
:param dict_name：键
:param properties: 字典
:param value: 值
:
return:
"""
if dict_name.find('.') > 0:
key = dict_name.split('.')[0]
properties.tdefault(key, {})
lf.__get_dict(dict_name[len(key) + 1:], properties[key], value) el:
properties[dict_name] = value
# 获取实例，保持单例
prop = PropertiesUtil()
if __name__ == "__main__":
# 调⽤⽅式，获取实例
# from util.ConfigUtil import prop
_config_dict("pdbc.properties"))
_config_value("pdbc.properties", "dbtype"))
⼤概是重点来了
有⼏点要先提⼀下。
1.我这⾥仅提供增删改查基本功能，其他诸如存储过程、函数等内容我⾃⼰也在探索中，故不列出。
2.使⽤importlib来实现动态加载，因为我不太喜欢开始就导⼊所有需要的模块，毕竟连接池不是⽤来同时连接所有类型数据库的。
3.PooledDB和建⽴连接时的config，我仅罗列了⼏项基本参数，更多的烦请⾃⾏查找资料。（这点是真的不好意思，因为本⼈只熟悉mysql相关内容，⽽且还很懒~~）
5.测试⽤的表名为TEST2，有两个字段，id 主键 数字类型，name 字符串类型。注意sqlrver的字符串请使⽤nvarchar类型，不然返回结果可能会乱码。（⾄于为什么不给建表语句的原因，算了，不编了，就是懒~~）
6. hba插⼊语句的参数使⽤：数字或者?作为占位符,hba的INSERT请使⽤UPSERT替换。且hba中'autocommit': True配置⼀定要，否则插⼊删除语句执⾏⽆效。
啰嗦了这么多，下⾯上代码。
# -*- coding:utf-8 -*-
"""
Description: DB⼯具类
@author: WangLeAi
@date: 2018/9/18
"""
from util.PropertiesUtil import prop
from DBUtils.PooledDB import PooledDB
import importlib
class DbPoolUtil(object):
def __init__(lf, config_file='config/pdbc.properties', db_type='mysql'):
"""
初始化
:param config_file: 配置⽂件地址
:param db_type: 数据库类型,⽀持 mysql, oracle, sqlrver, sqlite, hba
"""
properties_dic = _config_dict(config_file)
lf.__db_type = db_type
if lf.__db_type == "mysql":
config = {
'host': properties_dic['host_mysql'],
'port': int(properties_dic['port_mysql']),
'databa': properties_dic['databa_mysql'],
'ur': properties_dic['ur_mysql'],
'password': properties_dic['password_mysql'],
'chart': properties_dic['chart_mysql']
}
db_creator = importlib.import_module("pymysql")
lf.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
elif lf.__db_type == "oracle":
config = {
'ur': properties_dic['ur_orc'],
'password': properties_dic['password_orc'],
'dsn': "/".join(
[":".join([properties_dic['host_orc'], properties_dic['port_orc']]),
properties_dic['databa_orc']]),
'nencoding': properties_dic['nencoding_orc']
}
db_creator = importlib.import_module("cx_Oracle")
lf.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
elif lf.__db_type == "sqlrver":
config = {
'host': properties_dic['host_ms'],
'port': int(properties_dic['port_ms']),
'databa': properties_dic['databa_ms'],
'ur': properties_dic['ur_ms'],
'password': properties_dic['password_ms'],
'chart': properties_dic['chart_ms']
}
db_creator = importlib.import_module("pymssql")
lf.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config) elif lf.__db_type == "sqlite":
config = {
'databa': properties_dic['databa_sqlite3']
}
db_creator = importlib.import_module("sqlite3")
lf.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config) elif lf.__db_type == "hba":
# 'autocommit': True配置⼀定要，否则插⼊删除语句执⾏⽆效
config = {
'url': '{0}:{1}'.format(properties_dic['host_hb'], properties_dic['port_hb']),
'ur': properties_dic['ur_hb'],
'password': properties_dic['password_hb'],
'autocommit': True
}
db_creator = importlib.import_module("phoenixdb")
lf.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config) el:
rai Exception("unsupported databa type " + lf.__db_type)
def execute_query(lf, sql, dict_mark=Fal, args=()):
"""
执⾏查询语句，获取结果
:param sql:sql语句，注意防注⼊
:param dict_mark：是否以字典形式返回，默认为Fal
:param args：传⼊参数
:return：结果集
"""
result = []
conn = lf.__tion()
cur = conn.cursor()
try:
if dict_mark:
# name为description的第⼀个内容，表⽰为字段名
fields = [desc[0] for desc in cur.description]
rst = cur.fetchall()
if rst:
result = [dict(zip(fields, row)) for row in rst]
el:
result = cur.fetchall()
except Exception as e:
print('异常信息:' + str(e))
cur.clo()
conn.clo()
return result
def execute_query_single(lf, sql, dict_mark=Fal, args=()):
"""
执⾏查询语句，获取单⾏结果
:param sql:sql语句，注意防注⼊
:param dict_mark：是否以字典形式返回，默认为Fal
:param args：传⼊参数
:return：结果集
"""
