pooling 0.1.12

pooling anything.

pooling

pooling anything.

Install

pip install pooling

Usage Example 1

import MySQLdb
from MySQLdb.cursors import DictCursor

from pooling import PoolBase

class MysqlConnectionPool(PoolBase):
    
    def do_session_create(self, *create_args, **create_kwargs):
        create_kwargs.setdefault("cursorclass", DictCursor)
        create_kwargs.setdefault("autocommit", True)
        create_kwargs.setdefault("charset", "utf8mb4")
        return MySQLdb.connect(*create_args, **create_kwargs)

    def do_session_destory(self, real_session):
        real_session.close()

if __name__ == "__main__":
    conn_info = {
        "host": "127.0.0.1",
        "port": 3306,
        "user": "test",
        "password": "test",
        "db": "test",
    }
    pool = MysqlConnectionPool(pool_size=10, kwargs=conn_info)
    connection = pool.get_session()
    cursor = connection.cursor()
    count = cursor.execute("show databases")
    data = cursor.fetchall()
    print("rows count=", count)
    print("data=", data)
    assert count == len(data)

Usage Example 2

import MySQLdb
from MySQLdb.cursors import DictCursor

from pooling import Pool

def mysql_conn_create():
    conn_info = {
        "host": "127.0.0.1",
        "port": 3306,
        "user": "test",
        "password": "test",
        "db": "test",
    }
    conn = MySQLdb.connect(cursorclass=DictCursor, autocommit=True, **conn_info)
    return conn

def mysql_conn_close(session):
    session.close()

if __name__ == "__main__":
    pool = Pool(pool_size=10, create_factory=mysql_conn_create, destory_factory=mysql_conn_close)
    connection = pool.get_session()
    cursor = connection.cursor()
    count = cursor.execute("show databases")
    data = cursor.fetchall()
    print("rows count=", count)
    print("data=", data)
    assert count == len(data)

Usage Example 3

conn_info = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "test",
    "password": "test",
    "db": "test",
}

from pooling.mysql import MysqlConnectionPool

pool = MysqlConnectionPool(10, kwargs=conn_info)
connection = pool.get_session()
cursor = connection.cursor()
count = cursor.execute("show databases")
data = cursor.fetchall()
print("rows count=", count)
print("data=", data)
assert count == len(data)

Usage Example 4

conn_info = {
    "host": "127.0.0.1",
    "port": 3306,
    "user": "test",
    "password": "test",
    "db": "test",
}

from pooling.mysql import MysqlConnectionPool

pool = MysqlConnectionPool(10, kwargs=conn_info)
data = pool.query("select * from table1")
for row in data:
    print(row)

Note

• Call pool.get_session() returns a proxied connection instance.
• The returned proxied connection instance is a proxy instance of the real connection.
• When the returned proxied connection is being deleted, the real connection will be returned to the pool so that the real connection can be used again.
• A simple mysql connection pool can be imported by doing from pooling.mysql import MysqlConnectionPool. Compare with the MysqlConnectionPool implemented in Usage Example 1, it add ping() test and errors handler in get_session() method.
• The pooling.mysql module depends on mysqlclient, but pooling is NOT, so that mysqlclient is not auto installed after pooling installed. You should do pip install mysqlclient by yourself if you want to use MysqlConnectionPool.
• If you want to use MysqlConnectorPool, you need to install package mysql-connector-python by yourself.
• If you want to use PyMySQLConnectionPool, you need to install package PyMySQL by yourself.

Releases

v0.1.0

• First release.

v0.1.1

• MysqlConnectionPool.get_session() do connection.ping() to make sure the connection is avaiable.
• Pool.counter and Pool.version use thread safe counter.

v0.1.2

• Fix problem in using get_session with with statement. with pool.get_session() as session: pass got the real session instread of our wrapped Session.

v0.1.5

• Fix problem in handling a proxied real-session instance.
• Fix un-sleep-problem while connecting to mysql server failed.
• Add python2.7 support.

v0.1.6

• Use zenutils.importutils.
• Fix get_session timeout calc problem.

v0.1.7

• Unit test passed.

v0.1.8

• Add _pooling_is_connection_closed test.

v0.1.9

• Fix doing useless time.sleep problem in pooling.mysql.MysqlConnectionPoolBase.get_session.
• Add pooling.mysql.PyMySQLConnectionPool.

v0.1.10

• Fix PoolBase.return_session double called in with pool.get_session() as session: pass.
• PoolBase add ping_test support.

v0.1.11

• Doc update.

v0.1.12

• Doc update.