python3.13下使用数据库连接池（python连接数据库操作）
Python 3.13 Beta 1 来啦！
新特性一览：
- 交互式解释器升级，源自 PyPy，更智能，更懂你！
- 实验性 JIT 编译器，性能提升，代码运行更快！
- 激动人心的无 GIL 版本，多线程并发新体验！
- 垃圾回收更高效，增量式回收，内存管理更上一层楼！
- 新增类型注解，代码更规范，开发更安心！
也就是说从这个版本开始python开始了无GIL的时代，实现了真多线程并发。对于数据库开发的小伙伴来说很遗憾的是Cx_Oracle驱动包不再支持，使用dbutils的PooledDB来创建Oracle连接池不再可行；好消息是可以继续使用oracledb，而这个包更加高级，自带连接池。要适应新技术的发展，要对原来的代码进行改进。
- 对于项目数据库连接池的开发，必须使用单例模式。编制一个单例“原类”，继承此类的所有类都是单例。
import threading
class Single:
'''
单例模板类，继承该类子类均为单例模式
'''
_lock = threading.Lock() # python 的单例模式
def __new__(cls, *args, **kw):
while not hasattr(cls, "_instance"): # 适应多线程，双循环确认只创建1个实例:
with cls._lock:
while not hasattr(cls, "_instance"):
cls._instance = super().__new__(cls)
return cls._instance- 为了使用连接池，还必须保证所有的对象共享一个连接池，编一个基础MyDatabase类。
# import os
import pymysql
# import cx_Oracle as oracle #python3.10之后的版本不再支持
import oracledb as oracle #python3.10之后推荐使用，已经自带连接池
# oracle.init_oracle_client(lib_dir=r"C:\Program Files (x86)\Oracle\Instant Client\bin") # 针对oracle11.2的老版本需要采用这种厚模式
import threading
# from DBUtils.PooledDB import PooledDB # 使用数据库连接池1.3
from dbutils.pooled_db import PooledDB # 使用数据库连接池2.0,3.0以后对数据库参数大小写敏感，比如utf8不能写成UTF8
from abc import ABCMeta, abstractmethod
class MyDataBase(Single): # 继承一个单例类，就变成单例
'''
数据库基类，单例，必须被继承实现具体查寻操作
'''
__metaclass__ = ABCMeta
_pool = None # 连接池
_pool_lock = threading.Lock() # 连接池线程锁
@classmethod
def __init__(cls, db_config):
'''
单例模式建立连接池
'''
while cls._pool is None: # 适应多线程，双层循环确认pool对象没有被创建才创建，否则啥也不干
with cls._pool_lock: # 加线程锁
while cls._pool is None:
try:
cls.create_Pool(db_config) # 各类以自己的方式创建连接池
except Exception as e:
MyLogger.error(cls.__name__ + '未能正常创建连接池！请检查网络或服务器配制是否正确！' + str(e), __name__)
@classmethod
def getConnect(cls): # 基类获取链接的方法
'''
从连接池拿出一个链接并初始化光标
:return: 数据库链接和光标
'''
if cls._pool:
conn = cls._pool.connection()
cursor = conn.cursor()
return conn, cursor
return None, None- MySQL依旧使用PooledDB创建连接池，而oracledb自带连接池，并且oracledb获取链接的方式也不一样了。必须在子类各自实现连接池创建和链接获取方式。
class MySql(MyDataBase): # 继承一个单例类，就变成单例
"""MySQL数据库链接工具类"""
def __init__(self, databaseName='MySQL'):
'''
具体数据库类读取配制信息，并用父类初始化连接池
'''
db_config = self.loadDBconfig(databaseName) # 获取数据库配制信息
super().__init__(db_config)
@classmethod
def create_Pool(cls, db_config):
cls._pool = PooledDB(
creator=pymysql, # 对应的数据库驱动包
mincached=5, # 连接池里面初始的空闲链接
maxcached=10, # 最大的空闲链接数量
maxshared=10, # 最大的允许共享的连接数
maxusage=None, # 单个链接最大可以被复用的次数（默认未0/None, 代表不限制复用）
maxconnections=None, # 最大的允许的连接数（默认为0/None,代表多少链接都行）
blocking=True, # 确定超过最大连结束后的行为，true则阻塞等待有空闲链接，默认为报告错误
**db_config) # 数据库链接参数
class Oracle(MyDataBase):
os.environ['NLS_LANG'] = 'SIMPLIFIED CHINESE_CHINA.ZHS16GBK' # 解决cx_Oracle连接Oracle的乱码问题
def __init__(self, databaseName='Oracle1'):
'''
具体数据库类负责读取配制信息，并用父类初始化连接池
'''
db_config = self.loadDBconfig(databaseName)
oracle_path = ConfigUtil.getPath('oracle_dir')
# oracle.init_oracle_client(lib_dir=r"C:\Program Files (x86)\Oracle\Instant Client\bin")
oracle.init_oracle_client(lib_dir=oracle_path)
super().__init__(db_config)
@classmethod
def create_Pool(cls, db_config):
cls._pool = oracle.create_pool( # 以该类驱动包工具，创建连接池
min=8, # 连接池里面初始的空闲链接
max=16, # 最大的空闲链接数量
increment=1, # 自增数
**db_config) # 数据库链接参数
def getConnect(cls): # 由于oracledb 获取conn的方式不同，子类必须覆盖父类同名函数
if cls._pool:
conn = cls._pool.acquire()
cursor = conn.cursor()
return conn, cursor
return None, None