Python使用DBUtils连接部分主流数据库

需要了解的知识

首先呢，你需要了解下DBUtils，我的描述肯定没官网解释来的清晰，自行阅读后，你就会发现我为什么会选用PooledDB而不是其他作为连接池了。

其次，DBUtils支持所有遵循DP-API 2规范的数据库连接模块，也就是说除了我示例中所提供的几个数据库连接方式外，各位可以探索其他遵循此标准的连接模块，从而在此基础上拓展，成为连接更多种类数据库的通用工具类。

最后，以下内容均基于python3。

---------------------------------------------------------

10.23增补内容：支持hbase，更新字典返回方式以及部分方法扩展。

准备工作

首先，pip下支持DB-API 2规范的相关数据库连接模块和DBUtils。

pip install DBUtils
pip install pymysql（mysql）
pip install pymssql（sqlserver）
pip install cx_Oracle（oracle）
pip install phoenixdb（hbase）
pip install sqlite3（sqlite3 python自带）

其次，需要准备一份配置文件，姑且命名为pdbc.properties，以下是示例，根据数据库的连接信息进行修改即可。

# 数据库类型，支持mysql,oracle,sqlserver,sqlite3,hbase
# --------------------------------------------------------------
# mysql
# 连接数据库host
host_mysql=ip
# 连接数据库port
port_mysql=3306
# 连接数据库库名
database_mysql=dbname
# 用户
user_mysql=username
# 密码
password_mysql=password
# 字符集
charset_mysql=utf8
# --------------------------------------------------------------
# oracle
# 连接数据库host
host_orc=ip
# 连接数据库port
port_orc=1521
# 连接数据库库名
database_orc=dbname
# 用户
user_orc=username
# 密码
password_orc=password
# 字符集
nencoding_orc=utf8
# --------------------------------------------------------------
# sqlserver
# 连接数据库host
host_ms=ip
# 连接数据库port
port_ms=1433
# 连接数据库库名
database_ms=dbname
# 用户
user_ms=username
# 密码
password_ms=password
# 字符集
charset_ms=utf8
# --------------------------------------------------------------
# sqlite3
# 连接数据库文件名，sqlite不支持加密，不使用用户名和密码
database_sqlite3=path/to/your/dbname.db
# --------------------------------------------------------------
# hbase
# 连接数据库host
host_hb=ip
# 连接数据库port，phoenixdb连接使用8765端口而非2181等其他端口
port_hb=8765
# 用户
user_hb=username
# 密码
password_hb=password

然后，准备一份读取properties文件的工具类，姑且称为PropertiesUtil.py，可以寻找网上的，也可以参考我写的。

1.  
   # -*- coding:utf-8 -*-
2.  
    
3.  
    
4.  
   class PropertiesUtil(object):
5.  
   # 缓存配置
6.  
   __file_dict = {}
7.  
    
8.  
   def get_config_dict(self, file_path="pdbc.properties"):
9.  
   """
10.  
    获取资源文件，形成字典
11.  
    :param file_path: 文件路径
12.  
    :return:字典内容的key、value均为字符串
13.  
    """
14.  
    if file_path not in self.__file_dict:
15.  
    properties = {}
16.  
    with open(file_path, 'r', encoding='UTF-8') as pro_file:
17.  
    for line in pro_file.readlines():
18.  
    line = line.strip().replace(' ', '')
19.  
    if line.find('=') > 0 and not line.startswith('#'):
20.  
    strs = line.split('=')
21.  
    value = line[len(strs[0]) + 1:]
22.  
    self.__get_dict(strs[0].strip(), properties, value.strip())
23.  
    self.__file_dict[file_path] = properties
24.  
    return self.__file_dict[file_path]
25.  
     
26.  
    def get_config_value(self, file_path, prop_name):
27.  
    """
28.  
    获取资源文件，形成字典，获取属性值
29.  
    :param file_path: 文件路径
30.  
    :param prop_name: 属性名称
31.  
    :return: 返回字符串格式的属性值
32.  
    """
33.  
    return self.get_config_dict(file_path)[prop_name]
34.  
     
35.  
    def __get_dict(self, dict_name, properties, value):
36.  
    """
37.  
    递归获取配置字典
38.  
    :param dict_name:键
39.  
    :param properties: 字典
40.  
    :param value: 值
41.  
    :return:
42.  
    """
43.  
    if dict_name.find('.') > 0:
44.  
    key = dict_name.split('.')[0]
45.  
    properties.setdefault(key, {})
46.  
    self.__get_dict(dict_name[len(key) + 1:], properties[key], value)
47.  
    else:
48.  
    properties[dict_name] = value
49.  
     
50.  
     
51.  
    # 获取实例，保持单例
52.  
    prop = PropertiesUtil()
53.  
     
54.  
    if __name__ == "__main__":
55.  
    # 调用方式，获取实例
56.  
    # from util.ConfigUtil import prop
57.  
    print(prop.get_config_dict("pdbc.properties"))
58.  
    print(prop.get_config_value("pdbc.properties", "dbtype"))

大概是重点来了

有几点要先提一下。

1.我这里仅提供增删改查基本功能，其他诸如存储过程、函数等内容我自己也在探索中，故不列出。

2.使用importlib来实现动态加载，因为我不太喜欢开始就导入所有需要的模块，毕竟连接池不是用来同时连接所有类型数据库的。

3.PooledDB和建立连接时的config，我仅罗列了几项基本参数，更多的烦请自行查找资料。（这点是真的不好意思，因为本人只熟悉mysql相关内容，而且还很懒~~）

4.mysql和mssql语句的参数使用%s作为占位符，oracle和sqlite使用:数字作为占位符，sqllite还可以用?作为占位符，详情可以见代码中main函数示例。

5.测试用的表名为TEST2，有两个字段，id 主键 数字类型，name 字符串类型。注意sqlserver的字符串请使用nvarchar类型，不然返回结果可能会乱码。（至于为什么不给建表语句的原因，算了，不编了，就是懒~~）

6. hbase插入语句的参数使用:数字或者?作为占位符,hbase的INSERT请使用UPSERT替换。且hbase中'autocommit': True配置一定要，否则插入删除语句执行无效。

啰嗦了这么多，下面上代码。

1.  
   # -*- coding:utf-8 -*-
2.  
   """
3.  
   Description: DB工具类
4.  
    
5.  
   @author: WangLeAi
6.  
   @date: 2018/9/18
7.  
   """
8.  
   from util.PropertiesUtil import prop
9.  
   from DBUtils.PooledDB import PooledDB
10.  
    import importlib
11.  
     
12.  
     
13.  
    class DbPoolUtil(object):
14.  
    def __init__(self, config_file='config/pdbc.properties', db_type='mysql'):
15.  
    """
16.  
    初始化
17.  
    :param config_file: 配置文件地址
18.  
    :param db_type: 数据库类型,支持 mysql, oracle, sqlserver, sqlite, hbase
19.  
    """
20.  
    properties_dic = prop.get_config_dict(config_file)
21.  
    self.__db_type = db_type
22.  
    if self.__db_type == "mysql":
23.  
    config = {
24.  
    'host': properties_dic['host_mysql'],
25.  
    'port': int(properties_dic['port_mysql']),
26.  
    'database': properties_dic['database_mysql'],
27.  
    'user': properties_dic['user_mysql'],
28.  
    'password': properties_dic['password_mysql'],
29.  
    'charset': properties_dic['charset_mysql']
30.  
    }
31.  
    db_creator = importlib.import_module("pymysql")
32.  
    self.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
33.  
    elif self.__db_type == "oracle":
34.  
    config = {
35.  
    'user': properties_dic['user_orc'],
36.  
    'password': properties_dic['password_orc'],
37.  
    'dsn': "/".join(
38.  
    [":".join([properties_dic['host_orc'], properties_dic['port_orc']]),
39.  
    properties_dic['database_orc']]),
40.  
    'nencoding': properties_dic['nencoding_orc']
41.  
    }
42.  
    db_creator = importlib.import_module("cx_Oracle")
43.  
    self.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
44.  
    elif self.__db_type == "sqlserver":
45.  
    config = {
46.  
    'host': properties_dic['host_ms'],
47.  
    'port': int(properties_dic['port_ms']),
48.  
    'database': properties_dic['database_ms'],
49.  
    'user': properties_dic['user_ms'],
50.  
    'password': properties_dic['password_ms'],
51.  
    'charset': properties_dic['charset_ms']
52.  
    }
53.  
    db_creator = importlib.import_module("pymssql")
54.  
    self.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
55.  
    elif self.__db_type == "sqlite":
56.  
    config = {
57.  
    'database': properties_dic['database_sqlite3']
58.  
    }
59.  
    db_creator = importlib.import_module("sqlite3")
60.  
    self.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
61.  
    elif self.__db_type == "hbase":
62.  
    # 'autocommit': True配置一定要，否则插入删除语句执行无效
63.  
    config = {
64.  
    'url': 'http://{0}:{1}'.format(properties_dic['host_hb'], properties_dic['port_hb']),
65.  
    'user': properties_dic['user_hb'],
66.  
    'password': properties_dic['password_hb'],
67.  
    'autocommit': True
68.  
    }
69.  
    db_creator = importlib.import_module("phoenixdb")
70.  
    self.__pool = PooledDB(db_creator, maxcached=50, maxconnections=1000, maxusage=1000, **config)
71.  
    else:
72.  
    raise Exception("unsupported database type " + self.__db_type)
73.  
     
74.  
    def execute_query(self, sql, dict_mark=False, args=()):
75.  
    """
76.  
    执行查询语句，获取结果
77.  
    :param sql:sql语句，注意防注入
78.  
    :param dict_mark:是否以字典形式返回，默认为False
79.  
    :param args:传入参数
80.  
    :return:结果集
81.  
    """
82.  
    result = []
83.  
    conn = self.__pool.connection()
84.  
    cur = conn.cursor()
85.  
    try:
86.  
    if dict_mark:
87.  
    cur.execute(sql, args)
88.  
    # name为description的第一个内容，表示为字段名
89.  
    fields = [desc[0] for desc in cur.description]
90.  
    rst = cur.fetchall()
91.  
    if rst:
92.  
    result = [dict(zip(fields, row)) for row in rst]
93.  
    else:
94.  
    cur.execute(sql, args)
95.  
    result = cur.fetchall()
96.  
    except Exception as e:
97.  
    print('异常信息:' + str(e))
98.  
    cur.close()
99.  
    conn.close()
100.  
     return result
101.  
      
102.  
     def execute_query_single(self, sql, dict_mark=False, args=()):
103.  
     """
104.  
     执行查询语句，获取单行结果
105.  
     :param sql:sql语句，注意防注入
106.  
     :param dict_mark:是否以字典形式返回，默认为False
107.  
     :param args:传入参数
108.  
     :return:结果集
109.  
     """
110.  
     result = []
111.  
     conn = self.__pool.connection()
112.  
     cur = conn.cursor()
113.  
     try:
114.  
     if dict_mark:
115.  
     cur.execute(sql, args)
116.  
     # name为description的第一个内容，表示为字段名
117.  
     fields = [desc[0] for desc in cur.description]
118.  
     rst = cur.fetchone()
119.  
     if rst:
120.  
     result = dict(zip(fields, rst))
121.  
     else:
122.  
     cur.execute(sql, args)
123.  
     result = cur.fetchone()
124.  
     except Exception as e:
125.  
     print('异常信息:' + str(e))
126.  
     cur.close()
127.  
     conn.close()
128.  
     return result
129.  
      
130.  
     def execute_iud(self, sql, args=()):
131.  
     """
132.  
     执行增删改语句
133.  
     :param sql:sql语句，注意防注入
134.  
     :param args:传入参数
135.  
     :return:影响行数,mysql和sqlite有返回值
136.  
     """
137.  
     conn = self.__pool.connection()
138.  
     cur = conn.cursor()
139.  
     count = 0
140.  
     try:
141.  
     result = cur.execute(sql, args)
142.  
     conn.commit()
143.  
     if self.__db_type == "mysql":
144.  
     count = result
145.  
     if self.__db_type == "sqlite3":
146.  
     count = result.rowcount
147.  
     except Exception as e:
148.  
     print('异常信息:' + str(e))
149.  
     conn.rollback()
150.  
     cur.close()
151.  
     conn.close()
152.  
     return count
153.  
      
154.  
     def execute_many_iud(self, sql, args):
155.  
     """
156.  
     批量执行增删改语句
157.  
     :param sql:sql语句，注意防注入
158.  
     :param args:参数,内部元组或列表大小与sql语句中参数数量一致
159.  
     :return:影响行数，mysql和sqlite有返回值
160.  
     """
161.  
     conn = self.__pool.connection()
162.  
     cur = conn.cursor()
163.  
     count = 0
164.  
     loopK = 5000
165.  
     try:
166.  
     k = len(args)
167.  
     if k > loopK:
168.  
     n = k // loopK
169.  
     for i in range(n):
170.  
     arg = args[(i * loopK): ((i + 1) * loopK)]
171.  
     cur.executemany(sql, arg)
172.  
     conn.commit()
173.  
     arg = args[(n * loopK):]
174.  
     if len(arg) > 0:
175.  
     cur.executemany(sql, arg)
176.  
     conn.commit()
177.  
     else:
178.  
     result = cur.executemany(sql, args)
179.  
     conn.commit()
180.  
     if self.__db_type == "mysql":
181.  
     count = result
182.  
     if self.__db_type == "sqlite3":
183.  
     count = result.rowcount
184.  
     except Exception as e:
185.  
     print('异常信息:' + str(e))
186.  
     conn.rollback()
187.  
     cur.close()
188.  
     conn.close()
189.  
     return count
190.  
      
191.  
     def execute_proc(self, proc_name, args=()):
192.  
     """
193.  
     执行存储过程，mysql适用
194.  
     :param proc_name:存储过程/函数名
195.  
     :param args:参数
196.  
     :return:result为结果集，args_out为参数最终结果（用于out，顺序与传参一致）
197.  
     """
198.  
     result = ()
199.  
     args_out = ()
200.  
     conn = self.__pool.connection()
201.  
     cur = conn.cursor()
202.  
     try:
203.  
     cur.callproc(proc_name, args)
204.  
     result = cur.fetchall()
205.  
     if args:
206.  
     sql = "select " + ",".join(["_".join(["@", proc_name, str(index)]) for index in range(len(args))])
207.  
     cur.execute(sql)
208.  
     args_out = cur.fetchone()
209.  
     conn.commit()
210.  
     except Exception as e:
211.  
     print('异常信息:' + str(e))
212.  
     conn.rollback()
213.  
     cur.close()
214.  
     conn.close()
215.  
     return result, args_out
216.  
      
217.  
     def loop_row(self, obj, fun_name, sql, args=()):
218.  
     """
219.  
     执行查询语句，并且对游标每行结果反射调用某个处理方法
220.  
     主要是考虑一些表记录太大时，不能一次性取出,游标式取数据
221.  
     :param obj: 对象或者模块
222.  
     :param fun_name:调用方法名
223.  
     :param sql:sql语句，注意防注入
224.  
     :param args:传入参数
225.  
     :return:
226.  
     """
227.  
     conn = self.__pool.connection()
228.  
     cur = conn.cursor()
229.  
     try:
230.  
     cur.execute(sql, args)
231.  
     fun = getattr(obj, fun_name)
232.  
     while True:
233.  
     row = cur.fetchone()
234.  
     if row is None:
235.  
     break
236.  
     fun(row)
237.  
     except Exception as e:
238.  
     print('异常信息:' + str(e))
239.  
     cur.close()
240.  
     conn.close()
241.  
      
242.  
     def loop_row_custom(self, sql, args=()):
243.  
     """
244.  
     执行查询语句，并且对游标每行结果执行某些操作或者直接返回生成器
245.  
     主要是考虑一些表记录太大时，不能一次性取出,游标式取数据
246.  
     :param sql:sql语句，注意防注入
247.  
     :param args:传入参数
248.  
     :return:
249.  
     """
250.  
     conn = self.__pool.connection()
251.  
     cur = conn.cursor()
252.  
     try:
253.  
     cur.execute(sql, args)
254.  
     while True:
255.  
     row = cur.fetchone()
256.  
     if row is None:
257.  
     break
258.  
     # 在此编写你想做的操作
259.  
     print(row)
260.  
     except Exception as e:
261.  
     print('异常信息:' + str(e))
262.  
     cur.close()
263.  
     conn.close()
264.  
      
265.  
      
266.  
     # if __name__ == "__main__":
267.  
     # 使用demo，工作目录在项目目录的前提下,使用表为TEST2表
268.  
     # dbpool_util = DbPoolUtil(db_type="mysql")
269.  
     # sql1 = """DELETE FROM TEST2"""
270.  
     # result1 = dbpool_util.execute_iud(sql1)
271.  
     # print(result1)
272.  
      
273.  
     # mysql和mssql语句的参数使用%s作为占位符，oracle和sqlite使用:数字作为占位符(sqllite还可以用?作为占位符)
274.  
     # hbase插入语句的参数使用:数字或者?作为占位符,hbase的INSERT请使用UPSERT替换
275.  
     # sql2 = """INSERT INTO TEST2(id,name) VALUES (%s,%s)"""
276.  
     # sql2 = """INSERT INTO TEST2(id,name) VALUES (:1,:2)"""
277.  
     # sql2 = """UPSERT INTO TEST2(id,name) VALUES (?,?)"""
278.  
     # test_args2 = [(1, '王'), (2, '葬爱'), (3, 'shao'), (5, 'nian'), (8, 'wang')]
279.  
     # result2 = dbpool_util.execute_many_iud(sql2, test_args2)
280.  
     # print(result2)
281.  
      
282.  
     # sql3 = """SELECT id as wangleai,name as zangai FROM TEST2 """
283.  
     # result3 = dbpool_util.execute_query(sql3)
284.  
     # print(result3)
285.  
     # result3 = dbpool_util.execute_query_single(sql3)
286.  
     # print(result3)
287.  
     # result3 = dbpool_util.execute_query(sql3, dict_mark=True)
288.  
     # print(result3)
289.  
     # result3 = dbpool_util.execute_query_single(sql3, dict_mark=True)
290.  
     # print(result3)
291.  
     # dbpool_util.loop_row_custom(sql3)
292.  
      
293.  
     # 此处反射调用相关方法，文件就不给了，嫌麻烦
294.  
     # from util.ClassTest import ClsTest
295.  
     # cla_test = ClsTest()
296.  
     # dbpool_util.loop_row(cla_test, "print_row", sql3)
297.  
     #
298.  
     # import util.ModuleTest as mod_test
299.  
     #
300.  
     # dbpool_util.loop_row(mod_test, "print_row", sql3)
301.  
      
302.  
     # sql4 = """SELECT id,name FROM TEST2 where id = %s"""
303.  
     # sql4 = """SELECT id,name FROM TEST2 where id = :1"""
304.  
     # test_args4 = (3,)
305.  
     # result4 = dbpool_util.execute_query(sql4, args=test_args4)
306.  
     # print(result4)
307.  
      

以上，目前支持mysql，oracle，sqlserver，sqlite3和hbase。