Python数据库连接池基本用法 -- 2019-08-11 19:27:14

Posted 2022-07-02 gqy02

目录

• 基本用法
• 自制sqlhelper

原文: http://106.13.73.98/__/121/

@(Python数据库连接池)
确保已安装：pip install DBUtils
***

基本用法

先准备些数据

# 建了个表
create table userinfo(
    id int,
    name varchar(32),
    age int(3)
);

# 插入记录
insert into userinfo values      
    (1, 'user01', 21),     
    (2, 'user02', 22),     
    (3, 'user03', 23),
    (4, 'user04', 24);

创建使用数据库连接池

import pymysql
from DBUtils.PooledDB import PooledDB, SharedDBConnection

POOL = PooledDB(
    creator=pymysql,
    # 使用连接数据库的膜拜

    maxconnections=6,
    # 连接池允许的最大连接数，0和None表示不限制连接数

    mincached=2,
    # 初始化时，连接池中至少创建的空闲的连接，0表示不创建

    maxcached=5,
    # 连接池中最多闲置的连接，0和None表示不限制

    maxshared=3,
    # 连接池中最多共享的连接数量，0和None表示全部共享（这个参数不管指定多少，最后都会是0）
    # 因为pymysql和MySQLdb等模块的threadsafety都为1，所以值无论设置为多少，_maxcached永远为0，所以永远是所有连接都共享

    blocking=True,
    # 连接池中如果没有可用连接后，是否阻塞等待，True:等待 False:不等待，直接报错

    maxusage=None,
    # 一个连接最多可被使用的次数，None表示无限制

    setsession=[],
    # 开始会话前执行的命令列表，如:['set datastyle to ...', 'set time zone ...']

    ping=0,
    # ping MySQL服务端，检查服务是否可用，
    # 0 = None = never  从来没有
    # 1 = default = whenever it is reuqested  无论什么时候都要
    # 2 = when a cursor is created  创建游标时
    # 4 = when a query is executed  执行查询时
    # 7 = always  总是

    # 连接数据库：
    host='127.0.0.1',
    port=3306,
    user='tom',
    password='user@tom',
    database='blog116',
    charset='utf8',
)


# ====================================================================================================


def run():
    """
    检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常
    否则
    则优先去初始化时创建的链接中获取链接 SteadyDBConnection。
    然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。
    如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，再封装到PooledDedicatedDBConnection中并返回。
    一旦关闭链接后，连接就返回到连接池让后续线程继续使用。
    """

    conn = POOL.connection()
    cursor = conn.cursor(pymysql.cursors.DictCursor)

    # 开始sql语句
    cursor.execute('select * from userinfo')

    # 执行结果
    result = cursor.fetchall()
    [print(userinfo) for userinfo in result]

    # 这个不是关闭连接，而是将连接返回给连接池
    conn.close()


run()

运行后，我们将看到下面的打印信息：

***

自制sqlhelper

from DBUtils.PooledDB import PooledDB
import pymysql


class MySQLhelper(object):
    def __init__(self, host, port, dbuser, password, database):
        self.pool = PooledDB(
            creator=pymysql,
            # 使用连接数据库的膜拜

            maxconnections=6,
            # 连接池允许的最大连接数，0和None表示不限制连接数

            mincached=2,
            # 初始化时，连接池中至少创建的空闲的连接，0表示不创建

            maxcached=5,
            # 连接池中最多闲置的连接，0和None表示不限制

            maxshared=3,
            # 连接池中最多共享的连接数量，0和None表示全部共享（这个参数不管指定多少，最后都会是0）
            # 因为pymysql和MySQLdb等模块的threadsafety都为1，所以值无论设置为多少，_maxcached永远为0，所以永远是所有连接都共享

            blocking=True,
            # 连接池中如果没有可用连接后，是否阻塞等待，True:等待 False:不等待，直接报错

            maxusage=None,
            # 一个连接最多可被使用的次数，None表示无限制

            setsession=[],
            # 开始会话前执行的命令列表，如:['set datastyle to ...', 'set time zone ...']

            ping=0,
            # ping MySQL服务端，检查服务是否可用，
            # 0 = None = never  从来没有
            # 1 = default = whenever it is reuqested  无论什么时候都要
            # 2 = when a cursor is created  创建游标时
            # 4 = when a query is executed  执行查询时
            # 7 = always  总是

            # 连接数据库：
            host=host,
            port=int(port),
            user=dbuser,
            password=password,
            database=database,
            charset='utf8',
        )

    def create_conn_cursor(self):
        conn = self.pool.connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        return conn,cursor

    def fetch_all(self, sql, args):
        conn,cursor = self.create_conn_cursor()
        cursor.execute(sql,args)
        result = cursor.fetchall()
        cursor.close()
        conn.close()
        return result

    def insert_one(self,sql,args):
        conn,cursor = self.create_conn_cursor()
        res = cursor.execute(sql,args)
        conn.commit()
        conn.close()
        return res

    def update(self,sql,args):
        conn,cursor = self.create_conn_cursor()
        res = cursor.execute(sql,args)
        conn.commit()
        conn.close()
        return res


# 建立链接
sqlhelper = MySQLhelper('127.0.0.1', 3306, 'tom', 'user@tom', 'blog116')


# 开始sql语句

res = sqlhelper.fetch_all('select * from userinfo where id=%s', (3))

# res = sqlhelper.insert_one('insert into userinfo VALUES (%s, %s, %s)', (4, 'user04', 24))  # 执行成功返回1

# res = sqlhelper.update('update userinfo SET name=%s WHERE id=%s', ('user01', 1))  # 执行成功返回1


print(res)

原文: http://106.13.73.98/__/121/