21 lines
605 B
Python
21 lines
605 B
Python
import asyncio
|
|
import aiomysql
|
|
|
|
|
|
async def test_example(loop):
|
|
pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
|
|
user='root', password='',
|
|
db='mysql', loop=loop)
|
|
async with pool.acquire() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute("SELECT 42;")
|
|
print(cur.description)
|
|
(r,) = await cur.fetchone()
|
|
assert r == 42
|
|
pool.close()
|
|
await pool.wait_closed()
|
|
|
|
|
|
loop = asyncio.get_event_loop()
|
|
loop.run_until_complete(test_example(loop))
|