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)
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 theMysqlConnectionPool
implemented inUsage Example 1
, it add ping() test and errors handler in get_session() method. - The
pooling.mysql
module depends onmysqlclient
, butpooling
is NOT, so thatmysqlclient
is not auto installed afterpooling
installed. You should dopip install mysqlclient
by yourself if you want to useMysqlConnectionPool
. - If you want to use
MysqlConnectorPool
, you need to install packagemysql-connector-python
by yourself. - If you want to use
PyMySQLConnectionPool
, you need to install packagePyMySQL
by yourself.
Releases
v0.1.0
- First release.
v0.1.1
MysqlConnectionPool.get_session()
doconnection.ping()
to make sure the connection is avaiable.Pool.counter
andPool.version
use thread safe counter.
v0.1.2
- Fix problem in using
get_session
withwith
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.
Project details
Release history Release notifications | RSS feed
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.11.tar.gz
(7.4 kB
view hashes)