1.什么是dbutils
Dbutils是一套工具，可为数据库提供可靠，持久和汇总的连接，该连接可在各种多线程环境中使用。
2.使用代码记录
db_config.py 数据库配置类：
# -*- coding: UTF-8 -*-
import psycopg2
# 数据库信息
DB_TEST_HOST = "dev-pg.test.xxx.cloud"
DB_TEST_PORT = 1921
DB_TEST_DBNAME = "check_db"
DB_TEST_USER = "check_db"
DB_TEST_PASSWORD = "okk3py2323gBaX111"
# 数据库连接编码
DB_CHARSET = "utf8"
# mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)
DB_MIN_CACHED = 10
# maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)
DB_MAX_CACHED = 10
# maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用
DB_MAX_SHARED = 20
# maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)
DB_MAX_CONNECYIONS = 100
# blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配)
DB_BLOCKING = True
# maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)
DB_MAX_USAGE = 0
# setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...]
DB_SET_SESSION = None
# creator : 使用连接数据库的模块
DB_CREATOR = psycopg2
2.封装sqlHelper.py
import sys
import db_config as config
import psycopg2.extras
from dbutils.pooled_db import PooledDB
import threading
class PsycopgConn:
_instance_lock = threading.Lock()
def __init__(self):
self.init_pool()
def __new__(cls, *args, **kwargs):
if not hasattr(cls, '_instance'):
with PsycopgConn._instance_lock:
if not hasattr(cls, '_instance'):
PsycopgConn._instance = object.__new__(cls)
return PsycopgConn._instance
def get_pool_conn(self):
"""
获取连接池连接
:return:
"""
if not self._pool:
self.init_pool()
return self._pool.connection()
def init_pool(self):
"""
初始化连接池
:return:
"""
try:
pool = PooledDB(
creator=config.DB_CREATOR, # 使用连接数据库的模块 psycopg2
maxconnections=config.DB_MAX_CONNECYIONS, # 连接池允许的最大连接数，0 和 None 表示不限制连接数
mincached=config.DB_MIN_CACHED, # 初始化时，链接池中至少创建的空闲的链接，0 表示不创建
maxcached=config.DB_MAX_CACHED, # 链接池中最多闲置的链接，0 和 None 不限制
blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
maxusage=None, # 一个链接最多被重复使用的次数，None 表示无限制
setsession=[], # 开始会话前执行的命令列表
host=config.DB_TEST_HOST,
port=config.DB_TEST_PORT,
user=config.DB_TEST_USER,
password=config.DB_TEST_PASSWORD,
database=config.DB_TEST_DBNAME)
self._pool = pool
except:
print
'connect postgresql error'
self.close_pool()
def close_pool(self):
"""
关闭连接池连接
:return:
"""
if self._pool != None:
self._pool.close()
def SelectSql(self, sql):
"""
查询
:param sql:
:return:
"""
try:
conn = self.get_pool_conn()
cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) # 设置返回格式为字典
cursor.execute(sql)
result = cursor.fetchall()
except Exception as e:
print('execute sql {0} is error'.format(sql))
sys.exit('ERROR: load data from database error caused {0}'.format(str(e)))
finally:
cursor.close()
conn.close()
return result
def InsertSql(self, sql):
"""
插入数据
:param sql:
:return:
"""
try:
conn = self.get_pool_conn()
cursor = conn.cursor()
cursor.execute(sql)
result = True
except Exception as e:
print('ERROR: execute {0} causes error'.format(sql))
sys.exit('ERROR: update data from database error caused {0}'.format(str(e)))
finally:
cursor.close()
conn.commit()
conn.close()
return result
def UpdateSql(self, sql):
"""
更新数据
:param sql:
:return:
"""
try:
conn = self.get_pool_conn()
cursor = conn.cursor()
cursor.execute(sql)
result = True
except Exception as e:
print('ERROR: execute {0} causes error'.format(sql))
sys.exit('ERROR: update data from database error caused {0}'.format(str(e)))
finally:
cursor.close()
conn.commit()
conn.close()
return result
if __name__ == "__main__":
pgsql = PsycopgConn()
# 打开一个文件
with open('task.txt') as fr:
# 读取文件所有行
lines = fr.readlines()
lines = [i.rstrip() for i in lines]
list = []
list.append("taskId,batchId\n")
for taskId in lines:
sql = "select task_id, batch_id from ics_batch_info where batch_id=(select MAX(batch_id) from ics_batch_info WHERE task_id = '{taskId}')".format(
taskId=taskId)
print("执行sql: ", sql)
result = pgsql.SelectSql(sql)
if len(result) != 0:
list.append(result[0] + "," + str(result[1]) + "\n")
print(result)
list[len(list) - 1] = list[len(list) - 1].rstrip();
with open("最大批次查询结果.csv", 'w') as fw:
fw.writelines(list)
print("☺☺☺执行完毕☺☺☺")验证执行结果：
