"""
用于连接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