python oracle连接池_【Python + Oracle】Python Oracle连接池—改进版
Oracle 連接池第四版。
編程語言
Python
語言版本
3.5.3
實現功能
oracle 連接池，解決頻繁連接oracle數據庫帶來的連接資源耗費問題
程序托管
GitHub-OraclePool
修改日期
2020年8月3日
version 4.0 改進說明：
**[2020-04-21]**
+ 修復連接用完后，不放回連接池的`bug`
+ 修復使用類名稱獲取`pool`時的類名稱錯誤(`OrclConnPool`)，改為用`self`
+ 改進了對`config`連接信息的判斷邏輯，使得`config`的配置更加簡化
+ 改進了新的注釋，使用更加清晰
+ 改進了部分代碼的格式，使得其更規范
+ 新增`fetch_all`和`fetch_one`
+ 優化了`demo`內容和格式，使得其更規范
程序源碼：
# -*- coding: utf-8 -*-
"""
--------------------------------------
@File : oracle_pool.py
@Author : maixiaochai
@Created on : 2020/4/21 15:47
--------------------------------------
"""
import cx_Oracle as Oracle
from DBUtils.PooledDB import PooledDB
class OraclePool:
"""
1) 這里封裝了一些有關oracle連接池的功能;
2) sid和service_name，程序會自動判斷哪個有值，
若兩個都有值，則默認使用service_name；
3) 關于config的設置，注意只有 port 的值的類型是 int，以下是config樣例:
config = {
'user': 'maixiaochai',
'password': 'maixiaochai',
'host': '192.168.158.1',
'port': 1521,
'sid': 'maixiaochai',
'service_name': 'maixiaochai'
}
"""
def __init__(self, config):
"""
獲得連接池
:param config: dict Oracle連接信息
"""
self.__pool = self.__get_pool(config)
@staticmethod
def __get_pool(config):
"""
:param config: dict 連接Oracle的信息
---------------------------------------------
以下設置，根據需要進行配置
maxconnections=6, # 最大連接數，0或None表示不限制連接數
mincached=2, # 初始化時，連接池中至少創建的空閑連接。0表示不創建
maxcached=5, # 連接池中最多允許的空閑連接數，很久沒有用戶訪問，連接池釋放了一個，由6個變為5個，
# 又過了很久，不再釋放，因為該項設置的數量為5
maxshared=0, # 在多個線程中，最多共享的連接數，Python中無用，會最終設置為0
blocking=True, # 沒有閑置連接的時候是否等待， True，等待，阻塞住；False，不等待，拋出異常。
maxusage=None, # 一個連接最多被使用的次數，None表示無限制
setession=[], # 會話之前所執行的命令, 如["set charset ...", "set datestyle ..."]
ping=0, # 0 永遠不ping
# 1，默認值，用到連接時先ping一下服務器
# 2, 當cursor被創建時ping
# 4, 當SQL語句被執行時ping
# 7, 總是先ping
"""
dsn = None
host, port = config.get('host'), config.get('port')
if 'service_name' in config:
dsn = Oracle.makedsn(host, port, service_name=config.get('service_name'))
elif 'sid' in config:
dsn = Oracle.makedsn(host, port, sid=config.get('sid'))
pool = PooledDB(
Oracle,
mincached=5,
maxcached=10,
user=config.get('user'),
password=config.get('password'),
dsn=dsn
)
return pool
def __get_conn(self):
"""
從連接池中獲取一個連接，并獲取游標。
:return: conn, cursor
"""
conn = self.__pool.connection()
cursor = conn.cursor()
return conn, cursor
@staticmethod
def __reset_conn(conn, cursor):
"""
把連接放回連接池。
:return:
"""
cursor.close()
conn.close()
def __execute(self, sql, args=None):
"""
執行sql語句
:param sql: str sql語句
:param args: list sql語句參數列表
:param return: cursor
"""
conn, cursor = self.__get_conn()
if args:
cursor.execute(sql, args)
else:
cursor.execute(sql)
return conn, cursor
def fetch_all(self, sql, args=None):
"""
獲取全部結果
:param sql: str sql語句
:param args: list sql語句參數
:return: tuple fetch結果
"""
conn, cursor = self.__execute(sql, args)
result = cursor.fetchall()
self.__reset_conn(conn, cursor)
return result
def fetch_one(self, sql, args=None):
"""
獲取全部結果
:param sql: str sql語句
:param args: list sql語句參數
:return: tuple fetch結果
"""
conn, cursor = self.__execute(sql, args)
result = cursor.fetchone()
self.__reset_conn(conn, cursor)
return result
def execute_sql(self, sql, args=None):
"""
執行SQL語句。
:param sql: str sql語句
:param args: list sql語句參數
:return: tuple fetch結果
"""
conn, cursor = self.__execute(sql, args)
conn.commit()
self.__reset_conn(conn, cursor)
def __del__(self):
"""
關閉連接池。
"""
self.__pool.close()
def demo():
config = {
'user': 'maixiaochai',
'password': 'maixiaochai',
'host': '192.168.158.1',
'port': 1521,
'sid': 'maixiaochai',
'service_name': 'maixiaochai'
}
sql = "SELECT COUNT(*) FROM MAIXIAOCHAI"
orcl = OraclePool(config)
result = orcl.fetch_all(sql)
print(result)
if __name__ == "__main__":
demo()
version 3.0?改進說明：
*)修復一個 Bug，該 Bug導致連接池中有且僅有一個連接資源被使用，其余空閑。
*) 2019-12-27 12:45:21 by MaiXiaochai
version 2.0?改進說明：
*)將原來的 orcl_pool.py 文件單獨拿出來，創建新庫 OracleConnectionPool，便于編輯和分享；
*)更新"程序鏈接"內容和 URL。
*)2019-5-29 16:54:37 by MaiXiaochai
*)由原來的只能維持一個連接池改為可維持多個連接池；
*)加入了對 service_name 連接方式的支持；
*)添加了析構函數，對連接池資源進行回收；
*)對相關的代碼注釋進行了規范，使其看起來比較整潔；
*)添加了一個使用樣例。
*)2019-3-28 19:13:47 by MaiXiaochai
程序源碼：
version 3.0
# -*- coding: utf-8 -*-
# @File: oracle_conn_pool.py
# @Project: OraclePool
# @Date: 2019/5/29 15:19
# @Author: MaiXiaochai
# @Modify: 2020/1/6 16:02
import cx_Oracle as Oracle
from DBUtils.PooledDB import PooledDB
class OraclePool(object):
"""
1) 這里封裝了一些有關oracle連接池的功能;
2) sid和service_name，程序會自動判斷哪個有值，
若兩個都有值，則默認使用sid；
若只想用其中一個，則只需要把另一個設置為空即可。如，service_name = ''
3) 關于config的設置，注意只有 port 的值的類型是 int，以下是config樣例:
orcl_cfg = {
'user': 'user_name_str',
'passwd': 'passwd_str',
'host': 'xxx.xxx.xxx.xxx_str',
'port': port_int,
'sid': 'sid_str',
'service_name': 'service_name_str'}
"""
def __init__(self, config):
self.pool = OrclConnPool.__get_pool(config)
@staticmethod
def __get_pool(conf):
"""
一些 PoolDB 中可能會用到的參數，根據實際情況自己選擇
mincached： 啟動時開啟的空連接數量
maxcached： 連接池最大可用連接數量
maxshared： 連接池最大可共享連接數量
maxconnections： 最大允許連接數量
blocking： 達到最大數量時是否阻塞
maxusage： 單個連接最大復用次數
:param conf: dict 連接Oracle的信息
"""
host, port, sid, service_name = conf.get('host'), conf.get('port'), conf.get('sid'), conf.get('service_name')
dsn = None
if sid:
dsn = Oracle.makedsn(host, port, sid=sid)
elif service_name:
dsn = Oracle.makedsn(host, port, service_name=conf.get('service_name'))
__pool = PooledDB(Oracle, user=conf['user'], password=conf['passwd'], dsn=dsn, mincached=5, maxcached=30)
return __pool
def execute_sql(self, sql, args=None):
"""
執行sql語句
:param sql: str sql語句
:param args: list sql語句參數列表
"""
cur = self.pool.connection().cursor()
if args:
cur.execute(sql, args)
else:
cur.execute(sql)
def fetch_all(self, sql, args=None):
"""
獲取全部結果
:param sql: str sql語句
:param args: list sql語句參數
:return: tuple fetch結果
"""
cur = self.pool.connection().cursor()
if args:
cur.execute(sql, args)
else:
cur.execute(sql)
return cur.fetchall()
def __del__(self):
"""
在實例資源被回收時，關閉該連接池
"""
try:
self.pool.close()
except Exception:
pass
def simple_demo():
orcl_cfg = {
'user': 'hello',
'passwd': 'Python',
'host': '192.168.158.xxx',
'port': 1521,
'sid': '',
'service_name': 'MaiXiaochai'}
test_sql = "SELECT COUNT(1) FROM TEST_PYTHON"
orcl = OraclePool(orcl_cfg)
orcl.execute_sql(test_sql)
res = orcl.cur.fetchone()
print(res)
if __name__ == "__main__":
simple_demo()
傳送門：GitHub:OraclePool
The end.
總結
以上是生活随笔為你收集整理的python oracle连接池_【Python + Oracle】Python Oracle连接池—改进版的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 深扒Disruptor高性能的原因
- 下一篇： python list remove 删