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 details)
Built Distribution
File details
Details for the file pooling-0.1.11.tar.gz
.
File metadata
- Download URL: pooling-0.1.11.tar.gz
- Upload date:
- Size: 7.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/4.0.2 CPython/3.11.4
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 100e7569a75c509023b8bd5eb65d130ffc156211827539797f3e80de0ba5f02e |
|
MD5 | 32501586e6cb914ab48c2ba001feff27 |
|
BLAKE2b-256 | 81609c909498108203f1839681a2523ef704633a32828ae9f0de761d7c57e544 |
File details
Details for the file pooling-0.1.11-py3-none-any.whl
.
File metadata
- Download URL: pooling-0.1.11-py3-none-any.whl
- Upload date:
- Size: 7.5 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/4.0.2 CPython/3.11.4
File hashes
Algorithm | Hash digest | |
---|---|---|
SHA256 | 3e3a762832deb3d8a98ec07c5b027462369f5abd6b6183a030d1e5f0fea9b298 |
|
MD5 | 5acad30be463dc784906cdb9bc7f0a30 |
|
BLAKE2b-256 | 554ad4726abd0332da333ec9d772100dc5437b8eb4bee333d965d0b955762d55 |