aiomysql_async_connection_pool_select_query_example.py

1import asyncio
2import aiomysql
3
4async def test_example():
5    pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
6                                      user='root', password='',
7                                      db='mysql')
8    async with pool.acquire() as conn:
9        async with conn.cursor() as cur:
10            await cur.execute("SELECT 42;")
11            print(cur.description)
12            (r,) = await cur.fetchone()
13            assert r == 42
14            print(r)
15    pool.close()
16    await pool.wait_closed()
17
18asyncio.run(test_example())