python 全栈开发，Day121(DButils,websocket)

import pymysqlconn = pymysql.connect(host="localhost",port=3306,
user="root",password="",db="student",
charset="utf8")cur = conn.cursor(pymysql.cursors.DictCursor)# 插入一条数据
sql = "insert into stu(name,age) VALUE ('%s','%s')" %("小甜甜",22)cur.execute(sql)conn.commit()cur.close()conn.close()

import pymysql
from DBUtils.PooledDB import PooledDBPOOL = PooledDB(
creator=pymysql, # 使用链接数据库的模块
maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数
mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
maxcached=5, # 链接池中最多闲置的链接，0和None不限制
maxshared=1, # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
maxusage=None, # 一个链接最多被重复使用的次数，None表示无限制
setsession=[], # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
ping=0,
# ping MySQL服务端，检查是否服务可用。
# 如：0 = None = never,
# 1 = default = whenever it is requested,
# 2 = when a cursor is created,
# 4 = when a query is executed,
# 7 = always
host='127.0.0.1',
port=3306,
user='root',
password='',
database='book',
charset='utf8'
)

creator=pymysql,
host='127.0.0.1',
port=3306,
user='root',
password='',
database='book',
charset='utf8'

from utils import POOL
import pymysqldef func():
# 检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常
# 否则
# 则优先去初始化时创建的链接中获取链接 SteadyDBConnection。
# 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。
# 如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，再封装到PooledDedicatedDBConnection中并返回。
# 一旦关闭链接后，连接就返回到连接池让后续线程继续使用。
conn = POOL.connection() # 从连接池POOL中拿出一个已经创建好的连接,一次只能拿一个
cursor = conn.cursor(pymysql.cursors.DictCursor)
cursor.execute('select * from student')
result = list(cursor.fetchall()) # 使用list效率是很低的,这里仅做测试
print(result)
cursor.close()
conn.close()func()

[{'id': 1, 'name': '韩雪', 'age': 24, 'gender': '女'}, {'id': 2, 'name': '舒畅', 'age': 23, 'gender': '女'}, {'id': 3, 'name': '唐嫣', 'age': 25, 'gender': '女'}]

# -*- coding: UTF-8 -*-
import pymysql
from DBUtils.PooledDB import PooledDB
import DB_config as Config# import MySQLdb'''
@功能：PT数据库连接池
'''class PTConnectionPool(object):
__pool = None # def __init__(self):
# self.conn = self.__getConn()
# self.cursor = self.conn.cursor()
def __enter__(self):
self.conn = self.__getConn()
self.cursor = self.conn.cursor()
print("PT数据库创建con和cursor")
return self def __getConn(self):
if self.__pool is None:
self.__pool = PooledDB(creator=pymysql, mincached=Config.DB_MIN_CACHED, maxcached=Config.DB_MAX_CACHED,
maxshared=Config.DB_MAX_SHARED, maxconnections=Config.DB_MAX_CONNECYIONS,
blocking=Config.DB_BLOCKING, maxusage=Config.DB_MAX_USAGE,
setsession=Config.DB_SET_SESSION,
host=Config.DB_TEST_HOST, port=Config.DB_TEST_PORT,
user=Config.DB_TEST_USER, passwd=Config.DB_TEST_PASSWORD,
db=Config.DB_TEST_DBNAME, use_unicode=Config.DB_USE_UNICODE, charset=Config.DB_CHARSET) return self.__pool.connection() """
@summary: 释放连接池资源
""" def __exit__(self, type, value, trace):
self.cursor.close()
self.conn.close()
# print("PT连接池释放con和cursor") # 重连接池中取出一个连接
def getconn(self):
conn = self.__getConn()
# 设置返回数据为字典
cursor = conn.cursor(pymysql.cursors.DictCursor)
return cursor, conn # 关闭连接归还给连接池
# def close(self):
# self.cursor.close()
# self.conn.close()
# print u"PT连接池释放con和cursor"def getPTConnection():
return PTConnectionPool()

# -*- coding: UTF-8 -*-# TEST数据库信息
DB_TEST_HOST = "localhost"
DB_TEST_PORT = 3306
DB_TEST_DBNAME = "book"
DB_TEST_USER = "root"
DB_TEST_PASSWORD = ""# 数据库连接编码
DB_CHARSET = "utf8"# mincached : 启动时开启的闲置连接数量(缺省值 0 开始时不创建连接)
DB_MIN_CACHED = 10# maxcached : 连接池中允许的闲置的最多连接数量(缺省值 0 代表不闲置连接池大小)
DB_MAX_CACHED = 10# maxshared : 共享连接数允许的最大数量(缺省值 0 代表所有连接都是专用的)如果达到了最大数量,被请求为共享的连接将会被共享使用
DB_MAX_SHARED = 20# maxconnecyions : 创建连接池的最大数量(缺省值 0 代表不限制)
DB_MAX_CONNECYIONS = 100# blocking : 设置在连接池达到最大数量时的行为(缺省值 0 或 False 代表返回一个错误<toMany......> 其他代表阻塞直到连接数减少,连接被分配)
DB_BLOCKING = True# maxusage : 单个连接的最大允许复用次数(缺省值 0 或 False 代表不限制的复用).当达到最大数时,连接会自动重新连接(关闭和重新打开)
DB_MAX_USAGE = 0# setsession : 一个可选的SQL命令列表用于准备每个会话，如["set datestyle to german", ...]
DB_SET_SESSION = None# 是否使用unicode编码
DB_USE_UNICODE = True

# import MySQLdb
import pymysql
from MyDbUtils import getPTConnection
import DB_config as Configclass MysqlHelp(object): # mysql = None def __init__(self):
# self.connect()
self.db = getPTConnection() def __new__(cls, *args, **kwargs):
if not hasattr(cls, 'inst'):
cls.inst = super(MysqlHelp, cls).__new__(cls, *args, **kwargs)
return cls.inst # 查询所有
def selectall(self, sql='', param=()):
# 判断是否连接,并设置重连机制
# self.connected()
try:
cursor, conn = self.execute(sql, param)
res = cursor.fetchall()
self.close(cursor, conn)
return res
except Exception as e:
print('selectall except ', e.args)
self.close(cursor, conn)
return None # 查询一条
def selectone(self, sql='', param=()):
# self.connected()
try:
# cur = self.db.cursor()
cursor, conn = self.execute(sql, param)
res = cursor.fetchone()
self.close(cursor, conn)
return res
except Exception as e:
print('selectone except ', e.args)
self.close(cursor, conn)
return None # 增加
def insert(self, sql='', param=()):
# self.connected()
try:
# self.db.getconn().execute(sql, param)
cursor, conn = self.execute(sql, param)
# print('============')
# _id=self.db.conn.insert_id()
_id = cursor.lastrowid
# print('_id ', _id) conn.commit()
self.close(cursor, conn)
# 防止表中没有id返回0
if _id == 0:
return True
return _id
except Exception as e:
print('insert except ', e.args)
conn.rollback()
self.close(cursor, conn)
# self.conn.rollback()
return 0 # 增加多行
def insertmany(self, sql='', param=()):
# self.connected()
cursor, conn = self.db.getconn()
try:
cursor.executemany(sql, param)
# self.execute(sql,param)
conn.commit()
self.close(cursor, conn)
return True
except Exception as e:
print('insert many except ', e.args)
conn.rollback()
self.close(cursor, conn)
# self.conn.rollback()
return False # 删除
def delete(self, sql='', param=()):
# self.connected()
try:
cursor, conn = self.execute(sql, param)
conn.commit()
self.close(cursor, conn)
return True
except Exception as e:
print('delete except ', e.args) conn.rollback()
self.close(cursor, conn)
# self.conn.rollback()
return False # 更新
def update(self, sql='', param=()):
# self.connected()
try:
cursor, conn = self.execute(sql, param)
conn.commit()
self.close(cursor, conn)
return True
except Exception as e:
print('update except ', e.args)
conn.rollback()
self.close(cursor, conn)
# self.conn.rollback()
return False @classmethod
def getInstance(self):
if MysqlHelp.mysql == None:
MysqlHelp.mysql = MysqlHelp()
return MysqlHelp.mysql # 执行命令
# def execute(self, sql='', param=(), autoclose=False):
# cursor, conn = self.db.getconn()
# try:
# if param:
# cursor.execute(sql, param)
# else:
# cursor.execute(sql)
# conn.commit()
# if autoclose:
# self.close(cursor, conn)
# except Exception as e:
# pass
# return cursor, conn # 执行多条命令
'[{"sql":"xxx","param":"xx"}....]' # def executemany(self, list=[]):
# cursor, conn = self.db.getconn()
# try:
# for order in list:
# sql = order['sql']
# param = order['param']
# if param:
# cursor.execute(sql, param)
# else:
# cursor.execute(sql)
# conn.commit()
# self.close(cursor, conn)
# return True
# except Exception as e:
# print('execute failed========', e.args)
# conn.rollback()
# self.close(cursor, conn)
# return False # def connect(self):
# self.conn = pymysql.connect(user=Config.DB_TEST_USER, db=Config.DB_TEST_DBNAME, passwd=Config.DB_TEST_PASSWORD, host=Config.DB_TEST_HOST) def close(self, cursor, conn):
cursor.close()
conn.close()
# print("PT连接池释放con和cursor")

from mysqlhelp import MysqlHelp # 导入MysqlHelp类conn = MysqlHelp() # 实例化
sql = "select * from student" # SQL语句
res = conn.selectall(sql) # 执行sql,这里可以接收参数,用来拼接sql语句
print(res)

[{'id': 1, 'name': '韩雪', 'age': 24, 'gender': '女'}, {'id': 2, 'name': '舒畅', 'age': 23, 'gender': '女'}, {'id': 3, 'name': '唐嫣', 'age': 25, 'gender': '女'}]

from mysqlhelp import MysqlHelp # 导入MysqlHelp类conn = MysqlHelp() # 实例化
sql = "select * from student" # SQL语句
res = conn.selectone(sql) # 执行sql,这里可以接收参数,用来拼接sql语句
print(res)

from mysqlhelp import MysqlHelp # 导入MysqlHelp类conn = MysqlHelp() # 实例化
sql = "insert into student set name='{}',age='{}',gender='{}'".format('小甜甜','','女')
res = conn.insert(sql) # 执行sql
print(res)

from mysqlhelp import MysqlHelp # 导入MysqlHelp类conn = MysqlHelp() # 实例化
sql = "insert into student(name,age,gender) values (%s,%s,%s)"
res = conn.insertmany(sql,[('赵丽颖','','女'),('Angelababy','','女')]) # 执行sql,注意:多行是列表
print(res)

from mysqlhelp import MysqlHelp # 导入MysqlHelp类conn = MysqlHelp() # 实例化
sql = "update student set name='%s' where id=%s"
res = conn.update(sql,('杨颖','')) # 执行sql,这里可以接收参数,用来拼接sql语句
print(res)

from mysqlhelp import MysqlHelp # 导入MysqlHelp类conn = MysqlHelp() # 实例化
sql = "delete from student where id=%s"
res = conn.delete(sql,('')) # 执行sql,这里可以接收参数,用来拼接sql语句
print(res)