单例: 专业用来处理连接多的问题 (比如连接 Redis,zookeeper 等), 全局只有一个对象
单例代码
- def singleton(cls):
- instances = {}
- def _singleton(*args, **kwargs):
- if cls not in instances:
- instances[cls] = cls(*args, **kwargs)
- return instances[cls]
- return _singleton
实例代码
- from singleton import singleton
- #@singleton
- class MysqlOpers:
- def __init__(self):
- print('建立 mysql 连接')
- #伪代码 self.db = MySQLdb.connect()
- def select(self):
- pass
- m = MysqlOpers()
- n = MysqlOpers()
- c = MysqlOpers()
- print(id(m))
- print(id(n))
- print(id(c))
加上单例装饰器后
MySQL 连接池
- #coding=utf-8
- import traceback
- import MySQLdb
- from DBUtils.PooledDB import PooledDB
- db_pool_ins = None
- # 需要替换用户名, 密码等
- class DBPool():
- def __init__(self):
- self.pool = PooledDB(creator=MySQLdb, mincached=1, maxcached=10, maxconnections=100, blocking=True,
- host= "127.0.0.1", port=3306, user='', passwd='',
- db='test', charset='utf8',)
- def get_connection(self):
- return self.pool.connection()
- class DBAction():
- #连接池对象
- def __init__(self):
- #建立和数据库系统的连接
- global db_pool_ins
- if db_pool_ins == None:
- db_pool_ins = DBPool()
- self.conn = db_pool_ins.get_connection()
- #获取操作游标
- self.cursor = self.conn.cursor()
- def close_database(self):
- self.cursor.close()
- self.conn.close()
- def data_operate(self, sql, params=()):
- '''
- 数据的插入, 更新, 删除
- :param database:
- :param sql:
- :return: 成功: 0, 失败: 1
- '''
- try:
- self.cursor.execute(sql, params)
- self.conn.commit()
- return 0
- except:
- print("sql is %s, params is %s error. %s" % (sql, params, traceback.format_exc()))
- self.conn.rollback()
- raise Exception
- def data_operate_many(self, sql, params=()):
- '''
- 数据的插入, 更新, 删除
- :param sql:
- :param params:
- :return: 成功: 0, 失败: 1
- '''
- #执行 sql 语句
- self.cursor.executemany(sql, params)
- #提交到数据库执行
- self.conn.commit()
- def data_operate_count(self, sql, params=()):
- '''
- 数据的插入, 更新, 删除
- :return: 受影响的条数
- '''
- #执行 sql 语句
- count = self.cursor.execute(sql, params)
- #提交到数据库执行
- self.conn.commit()
- return count
- def data_inquiry(self, sql, size=10, params=()):
- '''
- :param database:
- :param sql:
- :return: ((),(),...,())
- '''
- self.cursor.execute(sql, params)
- result = self.cursor.fetchmany(size)
- return result
- def data_inquiry_all(self, sql, params=()):
- '''
- :param database:
- :param sql:
- :return: ((),(),...,())
- '''
- self.cursor.execute(sql, params)
- result = self.cursor.fetchall()
- return result
- def commit(self):
- self.conn.commit()
链接 MySQL 等关系型数据库, 应当采用此方式. 确保多连接存在.
来源: http://www.bubuko.com/infodetail-3089407.html