A thread safe sql executor for Python with connection pool. It helps you automatically manage database connections and transactions. It also provides ORM operations for single tables.
Project description
Usage Sample
import sqlexec
if __name__ == '__main__':
# sqlexec.init_db('test.db', driver='sqlite3', show_sql=True, debug=True)
# sqlexec.init_db("postgres://user:password@127.0.0.1:5432/testdb", driver='psycopg2', pool_size=5, show_sql=True, debug=True)
sqlexec.init_db(host='127.0.0.1', port='5432', user='xxx', password='xxx', database='testdb', show_sql=True, driver='psycopg2')
effected_rowcount = sqlexec.insert(table='person', name='zhangsan', age=15)
# if driver is 'pymysql' or 'mysql.connector' of MySQL, the select_key is 'SELECT LAST_INSERT_ID()'
select_key = "SELECT currval('person_id_seq')"
id = sqlexec.save(select_key=select_key, table='person', name='lisi', age=26)
id = sqlexec.save_sql(select_key, 'INSERT INTO person(name, age) VALUES(?,?)', 'wangwu', 38)
id = sqlexec.save_sql(select_key, 'INSERT INTO person(name, age) VALUES(:name, :age)', name='zhaoliu', age=45)
count = sqlexec.get('select count(1) from person')
# result: 4
persons = sqlexec.select('select id, name, age from person')
# result:
# (3, 'zhangsan', 15)
# (4, 'lisi', 26)
# (5, 'wangwu', 38)
# (6, 'zhaoliu', 45)
persons = sqlexec.select_one('select id, name, age from person where name = ?', 'zhangsan')
# result:
# (3, 'zhangsan', 15)
persons = sqlexec.select('select id, name, age from person where name = :name', name='zhangsan')
# result:
# [(3, 'zhangsan', 15)]
persons = sqlexec.query('select id, name, age from person')
# result:
# {'id': 3, 'name': 'zhangsan', 'age': 15}
# {'id': 4, 'name': 'lisi', 'age': 26}
# {'id': 5, 'name': 'wangwu', 'age': 38}
# {'id': 6, 'name': 'zhaoliu', 'age': 45}
persons = sqlexec.query_one('select id, name, age from person where name = ?', 'zhangsan')
# result:
# {'id': 3, 'name': 'zhangsan', 'age': 15}
persons = sqlexec.query('select id, name, age from person where name = :name', name='zhangsan')
# result:
# [{'id': 3, 'name': 'zhangsan', 'age': 15}]
effected_rowcount = sqlexec.execute('delete from person where id = ?', 5)
count = sqlexec.get('select count(1) from person')
# result: 3
effected_rowcount = sqlexec.execute('delete from person where id = :id', 4)
count = sqlexec.get('select count(1) from person')
# result: 2
Transaction
from sqlexec import with_transaction, transaction
@with_transaction
def test_transaction():
insert_func(....)
update_func(....)
def test_transaction2():
with transaction():
insert_func(....)
update_func(....)
If you want to operate MySQL database like Mybatis, may be you need MySqlx: https://pypi.org/project/mysqlx
If you want to operate PostgreSQL database like Mybatis, may be you need PgSqlx: https://pypi.org/project/pgsqlx
If you want to execute SQL like Mybatis, may be you need sqlx-batis: https://pypi.org/project/sqlx-batis
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
sqlx-exec-1.3.8.tar.gz
(13.3 kB
view hashes)