使用连接池封装pymysql

安装：pip install DBUtils==1.3

代码如下：

"""连接池操作pymysql"""
import pymysql
from DBUtils.PooledDB import PooledDB


class MysqlPool(object):

    def __init__(self):
        mysql_deploy = {
            "maxusage": None,
            "blocking": True,
            "mincached": 2,
            "maxcached": 5,
            "maxshared": 3,
            "maxconnections": 10,  # 连接池允许的最大连接数
            "creator": pymysql,
            "host": "xxx",
            "port": 3306,
            "user": "xxx",
            "password": "xxx",
            "database": "xxxl"
        }
        self.POOL = PooledDB(**mysql_deploy)

    def connect(self):
        """
        创建连接
        :return:
        """
        conn = self.POOL.connection()
        cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
        return conn, cursor

    def connect_close(self, conn, cursor):
        """
        关闭连接
        :param conn:
        :param cursor:
        :return:
        """
        cursor.close()
        conn.close()

    def fetch_one(self, sql_):
        """
        查单条数据
        :param sql_:
        :return:
        """
        conn, cursor = self.connect()
        cursor.execute(sql_)
        result = cursor.fetchone()
        self.connect_close(conn, cursor)
        return result

    def fetch_all(self, sql_):
        """
        批量数据
        :param sql_:
        :return:
        """
        conn, cursor = self.connect()
        cursor.execute(sql_)
        db_list = cursor.fetchall()
        self.connect_close(conn, cursor)
        return db_list

    def delect(self, sql_):
        conn, cursor = self.connect()
        row = cursor.execute(sql_)
        conn.commit()
        self.connect_close(conn, cursor)
        return row


if __name__ == '__main__':
    mysql = MysqlPool()
    sql = "DELETE FROM community.user_relation WHERE user_id=3725 OR to_user_id=3725;"
    data = mysql.delect(sql)
    print(data)

xxx   详细X

基本翻译

adj. 三十的

num. 罗马数字30

网络释义

xXx: 极限特工

rm xxx: 删除某个文件

xxx kisses: