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

怼烎@ 2023-02-09 14:22 36阅读 0赞

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

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

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

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

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

### 准备工作 ###

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

pip install DBTils
    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 settings
    
    mysql_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] =obj
            else:
                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 res
            except Exception as e:
                return None
            finally:
                self.close(conn, cursor)
    
        def selectAll(self, sql = '', *param):
            conn, cursor = self.createConn()
            try:
                self.executeParam(cursor, sql, param)
                res = cursor.fetchall()
                return res
            except Exception as e:
                return None
            finally:
                self.close(conn, cursor)
    
        def insert(self, sql, *param):
            conn, cursor = self.createConn()
            try:
                res = self.executeParam(cursor, sql, param)
                # 提交
                conn.commit()
                return res
            except Exception as e:
                # 回滚
                conn.rollback()
                return False
            finally:
                self.close(conn, cursor)
    
        def update(self, sql, *param):
            conn, cursor = self.createConn()
            try:
                res = self.executeParam(cursor, sql, *param)
                return res
            except Exception as e:
                conn.rollback()
                return False
            finally:
                self.close(conn, cursor)
    
        def delete(self, sql, *param):
            conn, cursor = self.createConn()
            try:
                res = self.executeParam(cursor, sql, param)
                return res
            except Exception as e:
                conn.rollback()
                return False
            finally:
                self.close(conn, cursor)

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

from until.base.mysql import mysqlHelper
    
    mySql = 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})

请求结果：

![20200521111621885.png][]

[https_www.cnblogs.com_zhuminghui_p_10930846.html]: https://www.cnblogs.com/zhuminghui/p/10930846.html
[20200521111621885.png]: https://img-blog.csdnimg.cn/20200521111621885.png