aiomysql

Posted 2022-06-05 yoyo1216

aiomysql：

import aiomysql
import asyncio


async def aiomysql_test():
    loop = asyncio.get_event_loop()
    # 第一种连接方法
    pool = await aiomysql.create_pool(host=‘127.0.0.1‘, port=3306, user=‘root‘, password=‘root‘, db=‘cfda‘, loop=loop,
                                      charset=‘utf8‘, autocommit=True)

    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            insert_sql = ‘insert into async_test_async(title) values("")‘.format(title)
            print(‘insert_sql:‘, insert_sql)
            # 插入数据
            await cursor.execute("insert into async_test_async(title) values(‘‘)".format(title))

            # 查询数据
            await cursor.execute("select * from async_test_async")
            data = await cursor.fetchall()
            print("data:", data)

            # 更新数据
            await cursor.execute("update async_test_async set title=‘‘ where id=".format(‘update‘, 10168))

            # 删除数据
            await cursor.execute("delete from async_test_async where id=".format(10174))

    # 第二种连接方法
    pool = await aiomysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, password=‘root‘, db=‘cfda‘, loop=loop,
                                  charset=‘utf8‘, autocommit=True)
    cursor = await pool.cursor()
    insert_sql = ‘insert into async_test_async(title) values("")‘.format(title)
    print(‘insert_sql:‘, insert_sql)
    # 插入数据
    await cursor.execute("insert into async_test_async(title) values(‘‘)".format(title))

    # 查询数据
    await cursor.execute("select * from async_test_async")
    data = await cursor.fetchall()
    print("data:", data)

    # 更新数据
    await cursor.execute("update async_test_async set title=‘‘ where id=".format(‘update‘, 10168))

    # 删除数据
    await cursor.execute("delete from async_test_async where id=".format(10174))

    cursor.close()
    pool.close()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(aiomysql_test())