pymysql-connection-manager
pymysql connection & pool manager for python3
Refactor pymysql connection
New features
- Parameter 'charset' default is utf8
- Parameter 'autocommit' default is True
- Added parameter 'timezone', default is '+00:00'
- Use pymysql.cursors.DictCursor by default
- Reconnect after the database connection is lost
- Add logs for creating connections, mysql warnings, exceptions, database queries, etc.
- Using the with...as syntax for transaction operations
- Provide simplified query methods such as fetch_all/fetch_row/fetch_column/fetch_first
- Provide simplified methods such as insert/insert_many/update/delete
1. Create pymysql connection
import pymysql
from pymysql_manager import Connection
db = Connection(host='192.0.0.1', database='foo', timezone='+8:00')2. Transaction
Before code:
try:
db.begin()
db.execute(....)
catch Exception:
db.rollback()
else:
db.commit()Now:
with db.transaction():
db.execute(...)3. Fetch rowsets
# executed: select * from foo where id between 5 and 10
all_rows = db.fetch_all('select * from foo where id between %s and %s', 5, 10)
# executed: select * from foo limit 1
first_row = db.fetch_row('select * from foo')
# executed: select * from foo limit 1
first_column_on_first_row = db.fetch_first('select * from foo')
# executed: select * from foo limit 1
third_column_on_first_row = db.fetch_column('select * from foo', column=3)4. Fetch by Iterator
When a result is large, it may be used SSCursor. But sometimes using limit ... offset ... can reduce the pressure on the database
by SSCursor
cursor = db.cursor(pymysql.cursors.SSCursor)
cursor.execute(sql)
while True:
row = cursor.fetchone()
if not row:
breakby fetch_iterator
for row in db.fetch_iterator(sql, per=1000, max=100000):
pass5. Single/Bulk Insert or Replace | Update | Delete
# insert ignore into mytable (foo, bar) values (1, 2)
db.insert('insert ignore into mytable', foo=1, bar=2)
# insert ignore into mytable (foo, bar) values (1, 2) on duplicate key update ...
db.insert('insert ignore into mytable on duplicate key update ...', **dict(foo=1, bar=2))
# insert ignore into mytable (id, name) values (1, 'foo'), (2, 'bar') on duplicate key update ...
db.insert_many('insert ignore into mytable on duplicate key update ...', ['id', 'name'], [(1, 'foo'), (2, 'bar')])
# update mytable set foo=1, bar=2 where id between %s and %s
db.update('update mytable where id between %s and %s', 10, 5, foo=1, bar=2)
db.update('update mytable where id between %s and %s', [10, 5], foo=1, bar=2)
db.update('update mytable where id between %s and %s', *[10, 5], **dict(foo=1, bar=2))
# update from mytable where id between %s and %s
db.delete('delete from mytable id between %s and %s', 10, 5)
db.delete('delete from mytable id between %s and %s', [10, 5])Connection Pool
1. Create connection pool
from pymysql_manager import ConnectionPooled
db = ConnectionPooled(host='192.0.0.1', database='foo',
pool_options=dict(max_size=10, max_usage=100000, idle=60, ttl=120))2. Execute SQL without the connection pool
db.execute(sql)
db.connection.execute(sql)3. Using connection pool to execute SQL
with db.pool() as connection:
connection.execute(sql)Connection Manager
1. Configuration
from pymysql_manager import ConnectionManager
db = ConnectionManager(default='foo',
foo=dict(host='192.0.0.1', database='foo', user='root', passwd=''),
bar=dict(host='192.0.0.1', database='bar', user='root', passwd=''))2. Get a connection
db.execute(sql) # use default connection
db['foo].execute(sql)
db.connection('foo').exeucte(sql)3. Get a connection from connection pool
with db.pool() as connection: pass # use default connection
with db['foo'].pool() as connection: pass
with db.connection('foo').pool() as connection: passLicense
The MIT License (MIT). Please see License File for more information.
