Python异步操作MySQL示例【使用aiomysql】_Python

Python异步操作MySQL示例【使用aiomysql】

2021-06-27 00:07Moxiao__墨箫 Python

这篇文章主要介绍了Python异步操作MySQL,结合实例形式分析了Python安装及使用aiomysql针对mysql数据库异步操作相关实现技巧,需要的朋友可以参考下

本文实例讲述了python异步操作mysql。分享给大家供大家参考，具体如下：

安装aiomysql

依赖

python3.4+
asyncio
pymysql

安装

				?

									pip install aiomysql

应用

基本的异步连接connection

				?

									import asyncio

									from aiomysql import create_pool

									loop = asyncio.get_event_loop()

									async def go():

									  async with create_pool(host='127.0.0.1', port=3306,

									              user='root', password='',

									              db='mysql', loop=loop) as pool:

									    async with pool.get() as conn:

									      async with conn.cursor() as cur:

									        await cur.execute("select 42;")

									        value = await cur.fetchone()

									        print(value)

									loop.run_until_complete(go())

异步的连接池 pool

				?

									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))

对象关系映射sqlalchemy - object relationship mapping

可以随意定义表结构，轻松调用查询、插入等操作方法。

				?

									import asyncio

									import sqlalchemy as sa

									from aiomysql.sa import create_engine

									metadata = sa.metadata()

									tbl = sa.table('tbl', metadata,

									        sa.column('id', sa.integer, primary_key=true),

									        sa.column('val', sa.string(255)))

									async def go(loop):

									  engine = await create_engine(user='root', db='test_pymysql',

									                 host='127.0.0.1', password='', loop=loop)

									  async with engine.acquire() as conn:

									    await conn.execute(tbl.insert().values(val='abc'))

									    await conn.execute(tbl.insert().values(val='xyz'))

									    async for row in conn.execute(tbl.select()):

									      print(row.id, row.val)

									  engine.close()

									  await engine.wait_closed()

									loop = asyncio.get_event_loop()

									loop.run_until_complete(go(loop))