【Sanic框架】第二章：基于DBUtils+pymysql的MySQL数据库连接

本文主要是介绍【Sanic框架】第二章：基于DBUtils+pymysql的MySQL数据库连接，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

在写这篇文章之前我并不知道竟然还有DBUtils这个东西，在我之前的理解中，一直以为只需要用pymysql连接数据库就行了（当然这也是可以的），用一次建立一次连接太差（大佬勿喷），后来经过同事的指点，了解到了DBUtils这个东西，但是不知道它的作用是什么？然后上网查找它的使用方法，大概了解它的作用和使用方法。

首先需要知道它的作用就像其他语言一样，连接数据库是需要一个连接池，而DBUtils就是建立连接池的一个包，如果了解python连接池的大佬略过。

它的主要作用就是连接池线程化，可以有效并安全的访问数据库，

这里只简单介绍它的两个外部接口PersistentDB和PooledDB,具体深入就不说了，推荐一个简单易懂的博客https://www.cnblogs.com/zhuminghui/p/10930846.html

PersistentDB：提供线程专用的数据库连接，并自动管理连接。
PooledDB ：提供线程间可共享的数据库连接，并自动管理连接。（目前使用的是这个接口）

准备工作

下载DBUtils和pymysql，别忘了记录到版本库文档中

pip install DBUTils
pip install pymysql

开始

准备工作做完了就开始码代码了,首先在配置文件中添加数据库配合信息

#/until/config/settings.py# 数据库连接信息
mysql_config = {"host": 'localhost', "user": 'user', "password": '123456', 'port': 3306, 'db': 'database'}

之后开始建立连接池并写增删改查的方法

import pymysql
from DBUtils.PooledDB import PooledDB
from until.config import settingsmysql_conf = settings.mysql_config# 数据库线程池单重态
'''
这里的作用就是不用反复建立数据库连接，想用的时候直接从连接池里拿就行了，连接池里有20个连接
'''
class Singleton(object):def __new__(cls, conf):db = conf.get('db', '')if hasattr(cls, '__obj'):if db not in cls.__obj:obj = super(Singleton, cls).__new__(cls)obj.POOL = PooledDB(pymysql, maxconnections=20, setsession=["SET GLOBAL time_zone = '+8:00'"], **conf)cls.__obj[db] =objelse:obj = super(Singleton, cls).__new__(cls)obj.POOL = PooledDB(pymysql, maxconnections=20, setsession=["SET GLOBAL time_zone = '+8:00'"], **conf)cls.__obj = {db: obj}return cls.__obj[db]class MysqlHelper(Singleton):def __new__(cls, conf):return super(MysqlHelper, cls).__new__(cls, conf)def createConn(self):# 创建一个连接conn = self.POOL.connection()# 得到一个可以执行SQL语句并且将结果作为字典返回的游标cursor = conn.cursor(pymysql.cursors.DictCursor)return conn, cursor# 执行def executeParam(self, cursor, sql, param):try:if param:return cursor.execute(sql, param)else:return cursor.execute(sql)except Exception as e:raise Exception(e.args[:])def close(self, conn, cursor):# 关闭游标cursor.close()# 释放连接conn.close()def selectOne(self, sql = '', *param):conn, cursor = self.createConn()try:self.executeParam(cursor, sql, param)res = cursor.fetchone()return resexcept Exception as e:return Nonefinally:self.close(conn, cursor)def selectAll(self, sql = '', *param):conn, cursor = self.createConn()try:self.executeParam(cursor, sql, param)res = cursor.fetchall()return resexcept Exception as e:return Nonefinally:self.close(conn, cursor)def insert(self, sql, *param):conn, cursor = self.createConn()try:res = self.executeParam(cursor, sql, param)# 提交conn.commit()return resexcept Exception as e:# 回滚conn.rollback()return Falsefinally:self.close(conn, cursor)def update(self, sql, *param):conn, cursor = self.createConn()try:res = self.executeParam(cursor, sql, *param)return resexcept Exception as e:conn.rollback()return Falsefinally:self.close(conn, cursor)def delete(self, sql, *param):conn, cursor = self.createConn()try:res = self.executeParam(cursor, sql, param)return resexcept Exception as e:conn.rollback()return Falsefinally:self.close(conn, cursor)

接下来就是使用了，我们来测试下，这里用到了dal文件夹，这个文件夹专门是用来写sql语句的，这里建一个hello_dal.py文件

from until.base.mysql import mysqlHelpermySql = mysqlHelper()def select():return mySql.selectAll(f"select * from user")

还是用之前的hello文件来测试，把代码改下

from routes import api
from dal import hello_dal
from sanic.response import json@api.route('/hello')
def select(request):res = hello_dal.select()return json({'res': res})

请求结果：