Tornado异步MySQL驱动程序
tormysql的Python项目详细描述
安装
pip install TorMySQL
二手龙卷风
示例池
from tornado.ioloop import IOLoop from tornado import gen import tormysql pool = tormysql.ConnectionPool( max_connections = 20, #max open connections idle_seconds = 7200, #conntion idle timeout time, 0 is not timeout wait_connection_timeout = 3, #wait connection timeout host = "127.0.0.1", user = "root", passwd = "TEST", db = "test", charset = "utf8" ) @gen.coroutine def test(): with (yield pool.Connection()) as conn: try: with conn.cursor() as cursor: yield cursor.execute("INSERT INTO test(id) VALUES(1)") except: yield conn.rollback() else: yield conn.commit() with conn.cursor() as cursor: yield cursor.execute("SELECT * FROM test") datas = cursor.fetchall() print datas yield pool.close() ioloop = IOLoop.instance() ioloop.run_sync(test)
示例助手
from tornado.ioloop import IOLoop from tornado import gen import tormysql pool = tormysql.helpers.ConnectionPool( max_connections = 20, #max open connections idle_seconds = 7200, #conntion idle timeout time, 0 is not timeout wait_connection_timeout = 3, #wait connection timeout host = "127.0.0.1", user = "root", passwd = "TEST", db = "test", charset = "utf8" ) @gen.coroutine def test(): tx = yield pool.begin() try: yield tx.execute("INSERT INTO test(id) VALUES(1)") except: yield tx.rollback() else: yield tx.commit() cursor = yield pool.execute("SELECT * FROM test") datas = cursor.fetchall() print datas yield pool.close() ioloop = IOLoop.instance() ioloop.run_sync(test)
单独使用异步
示例池
from asyncio import events import tormysql pool = tormysql.ConnectionPool( max_connections = 20, #max open connections idle_seconds = 7200, #conntion idle timeout time, 0 is not timeout wait_connection_timeout = 3, #wait connection timeout host = "127.0.0.1", user = "root", passwd = "TEST", db = "test", charset = "utf8" ) async def test(): async with await pool.Connection() as conn: try: async with conn.cursor() as cursor: await cursor.execute("INSERT INTO test(id) VALUES(1)") except: await conn.rollback() else: await conn.commit() async with conn.cursor() as cursor: await cursor.execute("SELECT * FROM test") datas = cursor.fetchall() print(datas) await pool.close() ioloop = events.get_event_loop() ioloop.run_until_complete(test)
示例助手
from asyncio import events import tormysql pool = tormysql.helpers.ConnectionPool( max_connections = 20, #max open connections idle_seconds = 7200, #conntion idle timeout time, 0 is not timeout wait_connection_timeout = 3, #wait connection timeout host = "127.0.0.1", user = "root", passwd = "TEST", db = "test", charset = "utf8" ) async def test(): async with await pool.begin() as tx: await tx.execute("INSERT INTO test(id) VALUES(1)") cursor = await pool.execute("SELECT * FROM test") datas = cursor.fetchall() print(datas) await pool.close() ioloop = events.get_event_loop() ioloop.run_until_complete(test)
资源
你可以阅读PyMySQL Documentation 在线获取更多信息。
许可证
tormysql使用mit许可,详细信息请参见许可文件。