jkklee/pymysql-pool

A simple but robust connection pool (with multiplexing) base on PyMySQL, mainly used for multi threads mode, which also compatible with single thread mode.

PyMySQL Connection Pool

中文文档

A simple but not simple mysql connection pool based on PyMySQL.

The problem to solve

While using pymysql with python multithreading, generally we will face the questions:

1. It can't share a connection created by main thread with all sub-threads. It will result in the following error:
   pymysql.err.InternalError: Packet sequence number wrong - got 0 expected 1
2. If we make every sub-thread to create a connection and close it when this sub-thread ends that's workable but obviously lead to high cost on establishing connections with MySQL.

Features

1. Simple: just use it, there is no extra learning costs.
2. Performance: almost no extra load compared to the original PyMysql(simple benchmark).
3. Flexible: pre_create connection or just create when really need; normal pool size and max pool size for the scalability, it all depends on you.
4. Thoughtful: connection lifetime and pre_ping mechanism, in case of borrow a brokend connection from the pool(such as closed by the mysql server due to wait_timeout setting).

Basic components

This module contains two classes:

• Connection class: this is a subclass of pymysql.connections.Connection. It can be used with or without a connection_pool, It used in the exact same way as pymysql. The details implementation of connection pool is hiddened (when used with a connection_pool additional actions are needed to maintain the pool).
• ConnectionPool class: instance of this class represents the actual connection_pool.

Misc

Using the concept of connection pool, there are also some aspects should be considered except the core features, such as:

• when getting connection from a pool: we should deal with the retry_num and retry_interval parameters，in order to give the borrower more chance and don't return the GetConnectionFromPoolError error directly.
• when putting connection back to pool: if the queries executed without exceptions, this connection can be putted back to the pool directly; but if exception occurred we have to decide whether this connection should be putted back to the pool depending on if it is reusable (depends on the exception type).

Luckily, this module will take care of these complicated details for you automaticly.

It also allows to create more than one connection_pool (with distinct ConnectionPool.name attribute) to be associated with different databases.

Usage example

Installation

pip install pymysql-pool

In the example below we're going to see how it works:

1. Create a pool with base/normal size is 2 and max size is 3, with pre_create_num=2 means will create 2 connections in the init phase:

   >>> import pymysqlpool
   >>> pymysqlpool.logger.setLevel('DEBUG')
   >>> config={'host':'xxxx', 'user':'xxx', 'password':'xxx', 'database':'xxx', 'autocommit':True}
   
   >>> mypool = pymysqlpool.ConnectionPool(size=2, maxsize=3, pre_create_num=2, name='mypool', **config)
   03-08 15:54:50    DEBUG: Create new connection in pool(mypool)
   03-08 15:54:50    DEBUG: Create new connection in pool(mypool)
   >>> mypool.total_num
   2
   
   >>> con1 = mypool.get_connection()
   12-25 21:38:48    DEBUG: Get connection from pool(mypool)
   >>> con2 = mypool.get_connection()
   12-25 21:38:51    DEBUG: Get connection from pool(mypool)
   >>> mypool.available_num
   0
   

2. Now the pool is empty, and we still borrow a connection from it, with the default parameters of get_connection(), we will see :

   >>> con3=mypool.get_connection()
   03-08 15:57:32    DEBUG: Retry to get connection from pool(mypool)
   03-08 15:57:32    DEBUG: Retry to get connection from pool(mypool)
   03-08 15:57:32    DEBUG: Retry to get connection from pool(mypool)
   03-08 15:57:33    DEBUG: Create new connection in pool(mypool)
   

   above message show us: although pool is empty, but the max size isn't reached, so after several times retry, a new connection is create(now max size of pool is reached)

3. Let's try to get another connection from pool:

   >>> con4=mypool.get_connection()
   03-08 16:29:43    DEBUG: Retry to get connection from pool(mypool)
   03-08 16:29:43    DEBUG: Retry to get connection from pool(mypool)
   03-08 16:29:43    DEBUG: Retry to get connection from pool(mypool)
   Traceback (most recent call last):
   File "/Users/kai/github/pymysql-pool/pymysqlpool.py", line 176, in get_connection
       conn = self._pool.pop()
   IndexError: pop from an empty deque
   
   ... ...
   
   pymysqlpool.GetConnectionFromPoolError: can't get connection from pool(mypool), retry_interval=0.1(s)
   

   we can see that after several times retry, finally raise a exception GetConnectionFromPoolError

4. Now let's see the connection's behavior while calling close() method or using it with Context Manager Protocol

   >>> con1.close()
   2017-12-25 21:39:56    DEBUG: Put connection back to pool(mypool)
   >>> with con2:
           with con2.cursor() as cur:
               cur.execute('select 1+1')
   
   1
   12-20 22:44:37    DEBUG: Put connection back to pool(mypool)
   >>> mypool.total_num
   3  # as we expect
   >>> mypool.available_num
   2  # as we expect
   

We can see that the module maintains the pool appropriately when (and only when) we call the close() method or use the Context Manager Protocol of the connection object.

Simple benchmark

I did a simple benchmark, focusing on the performance impact of the "extra" get and return operations in this module.
The test logic is in the simple-benchmark.py, You can check and do it yourself.
Below is my test(loop 50000 )

# 'pymysql-one-conn' is the best performing scenario, native pymysql, and all queries are done within a single connection
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-one-conn 50000
total 50000 finish within 6.564s.
7616.86 queries per second, avg 0.13 ms per query
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-one-conn 50000
total 50000 finish within 6.647s.
7522.31 queries per second, avg 0.13 ms per query
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-one-conn 50000
total 50000 finish within 6.558s.
7623.71 queries per second, avg 0.13 ms per query
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-one-conn 50000
total 50000 finish within 6.737s.
7421.67 queries per second, avg 0.13 ms per query

# 'pymysql-pool' uses connection pool (as long as the pool is greater than 1, it doesn't matter because the test logic is executed sequentially in a for loop).
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-pool 50000
total 50000 finish within 6.999s.
7143.77 queries per second, avg 0.14 ms per query
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-pool 50000
total 50000 finish within 7.066s.
7076.48 queries per second, avg 0.14 ms per query
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-pool 50000
total 50000 finish within 6.999s.
7143.71 queries per second, avg 0.14 ms per query
➜  pymysql-pool ✗ python3 simple-benchmark.py pymysql-pool 50000
total 50000 finish within 6.968s.
7175.65 queries per second, avg 0.14 ms per query

As we can see that one time get plus return operation only takes about 0.01ms.

Note

1. We should always use either the close() method or Context Manager Protocol of the connection object. Otherwise the pool will exhaust soon.

2. The Context Manager Protocol is preferred. It can achieve an effect similar to the "multiplexing", means the more Fine-Grained use of pool, also do more with less connections.

pymysql-pool's Issues

Question: Where can I find the code that handles the thread-safety in Connection checkin and checkout?

Thank you for your work.
I got curious how your pymysql-pool handles the thread-safety in Connection checkin and checkout?
Any Lock?
Thank you

why dont have a pool close method to close all connection?

why dont have a pool close method to close all connection?

problems using the Context Manager Protocol

Hi,
I am experiencing problems using the Context Manager Protocol. it seems doesn't release the connections, while manually it works fine.
Env:

pymysql-pool-0.3.5
Python 3.9.5
OS: "Raspbian GNU/Linux 10 (buster)" 

Using the following code and running 3 queries:

print("pool size before....:", ResourcesPool.size())
conn=ResourcesPool.get_connection()
logger.warning("Using Context_Manager_Protocol")
with conn as cur:
      data=cur.execute_query(query=query_string, args=field_values)
print("pool size after....:", ResourcesPool.size())

I get:

query-01
    pool size before....: 0
    pool size after....: 4
query-02
    pool size before....: 4
    pool size after....: 7
query-03
    pool size before....: 7
    pool size after....: 10

on the next query I get:

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/home/pi/.local/lib/python3.9/site-packages/pymysqlpool.py", line 184, in put_connection
    self._pool.put_nowait(conn)
  File "/usr/local/lib/python3.9/queue.py", line 191, in put_nowait
    return self.put(item, block=False)
  File "/usr/local/lib/python3.9/queue.py", line 137, in put
    raise Full
queue.Full

.... until stack is full

Instead using the following:

print("pool size before....:", ResourcesPool.size())
conn=ResourcesPool.get_connection()

try:
    cur=conn.cursor()
    cur.execute(query=query_string, args=field_values)
    data=cur.fetchall()
except (Exception) as e:
    logger.error(str(e))
    return None
    # raise
finally:
    conn.close()
print("pool size after....:", ResourcesPool.size())

I get:

query-01
    pool size before....: 0
    pool size after....: 1
query-02
    pool size before....: 1
    pool size after....: 1
query-03
    pool size before....: 1
    pool size after....: 1

Is there a my error?
Pool_jeekie_ISSUE.py.txt
Regards

Lost connection to MySQL server during query

Hello, I have bug in my app. I use this library with threading, so I have pool of 50 connections and 1-3 threading running at the same time. After a while my app was running, my threading can't pass data to DB. I decided to look at logs and I founded this error
[Thread-21][2020-07-29 07:10:10] (2013, 'Lost connection to MySQL server during query')

I think this is happened because I didn't use connection in some long time.
How I can fix it?

stubfile

Hi

Any plan to add stub file? Type checkers are complaining.
Thank you

a little problem with pool size but it doesn't matter

• NEW

class ConnectionPool:
    _HARD_LIMIT = 100
    _THREAD_LOCAL = threading.local()
    _THREAD_LOCAL.retry_counter = 0  # a counter used for debug get_connection() method

    def __init__(self, size=5, name=None, *args, **kwargs):
    *   self.size =  size < self._HARD_LIMIT and size or self._HARD_LIMIT
    *   self._pool = queue.Queue(self.size)
        self.name = name if name else '-'.join(
            [kwargs.get('host', 'localhost'), str(kwargs.get('port', 3306)),
             kwargs.get('user', ''), kwargs.get('database', '')])
    *   for _ in range(self.size):
            conn = Connection(*args, **kwargs)
            conn._pool = self
            self._pool.put(conn)

    def put_connection(self, conn):
        if not conn._pool:
            conn._pool = self
        conn.cursor().close()
        try:
            self._pool.put_nowait(conn)
            logger.debug("Put connection back to pool(%s)", self.name)
        except queue.Full:
    *       logger.warning("Put connection to pool(%s) error, pool is full, size:%d", self.name, self.size))

    # def size(self):
    #    return self._pool.qsize()

How do i connect to my Mysql server using SSHClient.

This is my code for the same.

Code:
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(**gb.ssh_config, look_for_keys=False)
ssh_channel = ssh_client.get_transport().open_channel(
'direct-tcpip',
('127.0.0.1', gb.dev_db_config["port"]),
(gb.dev_db_config["host"], gb.dev_db_config["port"]),
)
mysql_connection = pymysqlpool.ConnectionPool(
host='x.x.x.x',
port=3306,
user="admin",
password="Admin@123",
database='ai_netra'
)

Help me with this.

Not release connection from connection pool

Hi,

After get connection from connection pool, application run below code to make two database insert in a transaction.
The problem is when sql_2 execute fails, the database connection is released.
Would you please help suggest where is wrong.

with conn:
with conn.cursor() as cur:
sql_1= "INSERT INTO T1 (V1, V2) values(%s, %s)"
cur.execute(sql_1, ('Value1', 'Value2'))
sql_2= "INSERT INTO T2 (V3, V4) values(%s, %s)"
cur.execute(sql_2, ('Value3', 'Value4'))
conn.commit()

no arguments 'antocomit' in version 0.3.2

the version 0.3.2:
an unexpected keyword argument 'antocomit'
threr is no this arguments 'antocomit': True

Last version 0.4.6 doesn't honor cursorclass

My cursorclass setting is
pymysql.cursors.DictCursor
but I receive a tuple

Inconsistent results

When a record is inserted using one connection from the pool and then selected using a different connection from the pool. The second connection does not find the record.

execute_query() can't get the lastrowid when insert the record to DB

Use the function of execute_query() to insert the record to DB, can't get lastrowid.

And, the function of execute_query() use with context manager, so can't use same connection to execute conn.execute_query("SELECT LAST_INSERT_ID();") sentence to query the LAST_INSERT_ID().

hanging after the sql update command done

The script is hanging after done sql update command:
when I used Ctrl-Z to kill the process, I got the following message.
How can I solve the issue?
Exception ignored in: <module 'threading' from '/mnt/users/resops_user/services/py3.8/lib/python3.8/threading.py'>
Traceback (most recent call last):
File "/mnt/users/resops_user/services/py3.8/lib/python3.8/threading.py", line 1388, in _shutdown
lock.acquire()
KeyboardInterrupt:

Avoid Adding Log Handler To The Root Logger

Hi, shall we move the log handler from the root logger to the logger created in this module?

The current implementation calls logging.basicConfig, which will create a handler to root logger. In this approach, other loggers will print to stderr after importing this library because of the handler in the root logger.

We can simply resolve this issue by creating the handler in the logger instead of calling logging.basicConfig which adds to the root logger. How do you feel about this?

def _add_handler(logger: logging.Logger) -> None:
    handler = logging.StreamHandler()
    handler.setFomatter(
        logging.Formatter(
            fmt="%(asctime)s %(levelname)8s: %(message)s",
            datefmt="%m-%d %H:%M:%S",
        )
    )
    logger.addHandler(handler)

_add_handler(logger)

多线程情况下问题

我实现了连接池单例类，在多线程的情况下，会出现retry_counter没有定义的情况。

[enhancement] support singleton

pymysql.err.OperationalError: (2006, "MySQL server has gone away (BrokenPipeError(32, 'Broken pipe'))")

好像还是没解决session超时重连的问题，超过8个小时，还是会用之前的session

AttributeError: module 'pymysqlpool' has no attribute 'ConnectionPool'

I got this error:

Traceback (most recent call last):
File "/usr/src/app/./main.py", line 42, in
db_manager = DatabaseService(app.config)
File "/usr/src/app/./services/database_service.py", line 19, in init
self.pool = pymysqlpool.ConnectionPool(

on docker debian dist:

No LSB modules are available.
Distributor ID: Debian
Description: Debian GNU/Linux 12 (bookworm)
Release: 12
Codename: bookworm

But in Windows 10 it works. There are the same version of python and same version of the packages.

import pymysql.cursors
import pymysqlpool
import time
from log.log import Log

Inicializando el logger

logger = Log.get_logger()

class DatabaseService:
def init(self, config):
"""
Constructor de la clase DatabaseManager. Inicializa la conexión
al pool de conexiones con la configuración proporcionada.

    :param config: Diccionario con la configuración de la base de datos.
    """
    self.pool = pymysqlpool.ConnectionPool(
        host=config['MYSQL_HOST'],
        user=config['MYSQL_USER'],
        password=config['MYSQL_PASSWORD'],
        db=config['MYSQL_DB'],
        cursorclass=pymysql.cursors.DictCursor,
        autocommit=True,
        size=5,
        maxsize=7,
        pre_create_num=2,
        name='pool1'
    )
    self.connection = None

def __enter__(self):
    """
    Método especial para permitir el uso de DatabaseManager
    dentro de un bloque with.

    :return: Una conexión a la base de datos.
    """
    self.connection = self.get_connection_with_retry()
    return self.connection

def __exit__(self, exc_type, exc_val, exc_tb):
    """
    Método especial que se llama cuando se sale del bloque with,
    asegurando que la conexión se cierre.

    :param exc_type: Tipo de excepción.
    :param exc_val: Valor de la excepción.
    :param exc_tb: Traceback de la excepción.
    """
    self.close()

def get_connection_with_retry(self, max_retries=5, delay=1):
    """
    Intenta obtener una conexión del pool, reintentando hasta
    max_retries veces con un retraso de delay segundos entre intentos.

    :param max_retries: Número máximo de reintentos.
    :param delay: Retraso entre reintentos en segundos.
    :return: Una conexión a la base de datos.
    """
    retry_count = 0
    while retry_count < max_retries:
        try:
            return self.pool.get_connection()
        except pymysqlpool.GetConnectionFromPoolError:
            retry_count += 1
            time.sleep(delay)
    logger.error(f"Can't get connection after {max_retries} retries")
    raise pymysqlpool.GetConnectionFromPoolError(f"Can't get connection after {max_retries} retries")

def close(self):
    """
    Cierra la conexión actual a la base de datos si existe.

    """
    if self.connection is None:
        logger.warning("No connection to close.")
    else:
        try:
            self.connection.close()
            self.connection = None
            logger.info("Connection closed successfully.")
        except Exception as e:
            logger.error(f"Error closing connection: {e}")