pip install pymysql
pip install DBUtils

import pymysql

conn = 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 PooledDB

POOL = 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，所以永远是所有连接都共享。
    blocaking=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‘
)

from utils import POOL
import pymysql


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


func()

import pymysql
from DBUtils.PooledDB import PooledDB
import DB_config as Config

"""功能：PT数据库连接池"""


class PTConnectionPool(object):
    __pool = None

    def __enter__(self):
        self.conn = self.__getConn()
        self.cursor = self.conn.cursor()
        print(‘PT数据库创建conn和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_POST, 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连接池释放conn和cursor‘)

    # 重连接池中取出一个连接
    def getconn(self):
        conn = self.__getConn()
        # 设置返回数据为字典
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        return cursor, conn


def getPTConnection():
    return PTConnectionPool()

# TEST数据库信息
DB_TEST_HOST = ‘192.168.37.131‘
DB_TEST_POST = ‘3306‘
DB_TEST_DBNAME = ‘book‘
DB_TEST_USER = ‘root‘
DB_TEST_PASSWORD = ‘123456‘

# 数据库连接编码
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