PyMySQL Pool

raw JSON →
0.5.0 verified Sat May 09 auth: no python

PyMySQL Pool is a high-performance MySQL connection pool built on top of PyMySQL, supporting connection reuse, configurable pool limits, connection lifetime management, and automatic rollback/re-autocommit on return. Version 0.5.0 (stable) supports Python >=3.7, last release 2022-06.

pip install pymysql-pool
error AttributeError: module 'pymysqlpool' has no attribute 'ConnectionPool'
cause Outdated version or incorrect import (e.g. from pymysqlpool.pool import ConnectionPool). The top-level import is the correct one only from v0.4.x onward.
fix
Use 'from pymysqlpool import ConnectionPool' after pip install --upgrade pymysql-pool.
error pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')
cause The server-side wait_timeout closes idle connections. pymysql-pool's con_lifetime parameter was introduced to re-create connections automatically.
fix
Set con_lifetime (in seconds) to a value less than MySQL wait_timeout, e.g. pool = ConnectionPool(..., con_lifetime=3600).
error TypeError: __init__() got an unexpected keyword argument 'max_size'
cause Parameter renamed from max_size to maxsize in v0.3.7.
fix
Use maxsize=... instead.
breaking In v0.3.7, the parameter 'max_size' was renamed to 'maxsize' in ConnectionPool.__init__. Older code using max_size will raise TypeError.
fix Use maxsize=... instead of max_size=...
breaking In v0.4.3, pool.size was renamed to pool.available_num and pool.connection_num to pool.total_num.
fix Use pool.available_num and pool.total_num instead.
gotcha Connections must be returned to the pool via pool.release(conn). Forgetting to release may exhaust the pool and cause hangs.
fix Always call pool.release(conn) in a finally block or use context manager (pool.connection()) if available.

Create a pool, borrow a connection via get_conn(), use it, then release it back.

from pymysqlpool import ConnectionPool

pool = ConnectionPool(
    host='localhost',
    user='root',
    password=os.environ.get('MYSQL_PWD', ''),
    database='test',
    maxsize=10,
    pre_create_num=2,
    pool_name='mypool'
)
conn = pool.get_conn()
cursor = conn.cursor()
cursor.execute('SELECT 1')
result = cursor.fetchone()
cursor.close()
pool.release(conn)
print(result)