Skip to main content

pooling anything.

Project description

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)

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.

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.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

pooling-0.1.1.tar.gz (5.0 kB view hashes)

Uploaded Source

Built Distribution

pooling-0.1.1-py3-none-any.whl (5.3 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page