pymysql-connection-manager

pymysql connection & pool manager for python3

Refactor pymysql connection

New features

1. Parameter 'charset' default is utf8
2. Parameter 'autocommit' default is True
3. Added parameter 'timezone', default is '+00:00'
4. Use pymysql.cursors.DictCursor by default
5. Reconnect after the database connection is lost
6. Add logs for creating connections, mysql warnings, exceptions, database queries, etc.
7. Using the with...as syntax for transaction operations
8. Provide simplified query methods such as fetch_all/fetch_row/fetch_column/fetch_first
9. 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:
    break

by fetch_iterator

for row in db.fetch_iterator(sql, per=1000, max=100000):
  pass

5. 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: pass

License

The MIT License (MIT). Please see License File for more information.