"""
用於連接Oracle數據庫，使用前需確認系統已安裝cx_Oracle庫和oracle客戶端
cx_Oracle就可以在PyPI中下載,https://pypi.python.org/pypi，根據Python版本和系統位數進行下載
Oracle客戶端，百度找instantclient，根據系統確定好是下載32位的還是64位
"""
# -*- coding:utf-8 -*-
import cx_Oracle as Oracle
from DBUtils.PooledDB import PooledDB
class OrclPool(object):
"""
1) 這里封裝了一些有關oracle連接池的功能;
2) sid和service_name，程序會自動判斷哪個有值，
若兩個都有值，則默認使用sid；
若只想用其中一個，則只需要把另一個設置為空即可。如，service_name = ''
3) 關於config的設置，只有 port 的值的類型是 int，其他均為str:
"""
def __init__(self, config):
self.conn = OrclPool.__get_conn(config) # 獲得連接池，Oracle連接信息
self.cur = self.conn.cursor() # 獲取操作游標
@staticmethod
def __get_conn(conf):
"""
: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
"""
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=1, maxcached=5)
return __pool.connection()
def execute_sql(self, sql, args=None):
"""
執行sql語句
:param sql: str sql語句
:param args: list sql語句參數列表
"""
if args:
self.cur.execute(sql, args)
else:
self.cur.execute(sql)
def fetch_all(self, sql, args=None):
"""
獲取全部結果
:param sql: str sql語句
:param args: list sql語句參數
:return: tuple fetch結果
"""
self.execute_sql(sql, args) # 執行 SQL 語句
return self.cur.fetchall() # 返回執行的結果
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 commit_sql(self, sql):
self.cur.execute(sql)
self.conn.commit() # 用於SQL語句提交事務
def __del__(self):
"""
在實例資源被回收時，關閉該連接池與數據庫
"""
self.cur.close()
self.conn.close()
if __name__ == "__main__":
orcl_cfg = {
'user': 'username', # 數據庫用戶名字
'passwd': 'password', # 數據庫密碼
'host': '0.0.0.0', # 數據庫鏈接地址
'port': 3000, # 端口
'sid': '',
'service_name': 'TESTDB', # 數據庫名字
}
orcl = OrclPool(orcl_cfg)
部分內容參考：https://blog.csdn.net/maixiaochai/article/details/82986517