Python学习之单例模式 单例模式的代码实现

python编程中，尤其是web开发，访问数据库这种比较单一的资源，必须采用单例模式，利用连接池建立对数据库资源的连接，提高程序效率，限制内存使用量，避免系统崩溃。

import pymysql													# 访问MySQL数据库
import cx_Oracle as oracle				# 访问Oracle数据库
import pandas as pd									# 使用pandas数据分析

import threading

class Single:
    '''
		单例模板类，继承该类的子类，均为单例模式
    '''
    _lock = threading.Lock()                        # 线程锁
    
    def __new__(cls, *args, **kw):
        while not hasattr(cls, "_instance"): 
            with cls._lock:																	# 线程锁，实现线程安全
                while not hasattr(cls, "_instance"):
                    cls._instance = super().__new__(cls, *args, **kw)
                    print(cls.__name__ + '创建对象')
        return cls._instance

class MyDataBase(Single):  # 继承一个单例类，就变成单例
    '''
    数据库通用基类，实现数据库操作一些通用功能
    '''
   
    _pool = None                                          # 声明连接池
    
    @classmethod
    def getConnect(cls):
        if cls._instance._pool:
            conn = cls._instance._pool.connection()				#从连接池取连接
            cursor = conn.cursor()
            return conn, cursor
        return None, None
	
    def getDataFrame(self, sql, parse_dates=None, index_col=None):
          '''
          使用pandas读取数据库数据，返回datafram、pandas对象，方便数据分析和绘图
          '''
          conn, _ = self.getConnect()
          if conn is None: return None
          try:
              df = pd.read_sql(sql, conn, parse_dates=parse_dates, index_col=index_col, coerce_float=True)
              return df, pd
          except Exception as e:
              print('Pandas 读取数据库失败'+ str(e))
          finally:
              if conn: conn.close()
	
# 实现with as 必须重载 __enter__、__exit__ 两个函数
 		def __enter__(cls):

        cls.conn, cls.cursor = cls.getConnect()
        return cls.conn, cls.cursor
    
    def __exit__(cls, exc_type, exc_val, exc_tb):

        if cls.cursor:
            cls.cursor.close()
            cls.cursor = None
        if cls.conn:
            cls.conn.close()
            cls.conn = None

class MySql(MyDataBase):  # 继承一个单例类，就变成单例
    """MySQL数据库链接类"""

   def __init__(self):										# 饿汉模式，立即创建单例连接池
        while MySql._pool is None:
            with MySql._lock:					# 线程安全，创建连接池对象单例
                while MySql._pool is None:
                    try:
                        MySql._pool = PooledDB(pymysql, mincached=5, maxcached=15, maxshared=10, maxusage=50000,
                                             host="127.0.0.1",port=3306, user='root', passwd='123456',db='xx_db', connect_timeout=5)
                    except Exception as e:
                        print('MySQL 连接池创建错误！' + str(e))

class Oracle(MyDataBase):

    def __init__(self):
        while Oracle._pool is None:
            with Oracle._lock:			# 线程安全，创建连接池对象单例
                while Oracle._pool is None:
                    try:
                        Oracle._pool = PooledDB(oracle, mincached=5, maxcached=15, maxshared=10, maxusage=50000,
                                              user='zhang3', password='abcd123',dsn='127.0.0.1:1521/orcl')
                    except Exception as e:
                        print('Oracle 连接池创建错误！' + str(e))


if __name__ == '__main__':
		'''
		使用举例
		'''
		mysql_db = MySql()
		sql = 'select * from table1 t;'
		df, pd = mysql_db.getDataFrame(sql, parse_dates='日期', index_col='日期')
  	df.head()

		with MySql() as conn_curs:
				conn, curs = conn_curs
				sql = 'select * from table1 t;'
				rs = curs.execute(sql)
				result = curs.fetchall()
        for rows in result:
            print(rows)

		oracle_db = Oracle()
		sql = 'select * from table t'
		df, pd = mysql_db.getDataFrame(sql, parse_dates='日期', index_col='日期')
  	df.head()