您的位置：首页 > 脚本大全 > > 正文

pythonmysql使用教程（Python异步操作MySQL示例使用aiomysql）

更多 时间：2021-09-29 03:09:16 类别：脚本大全 浏览量：1059

pythonmysql使用教程

Python异步操作MySQL示例使用aiomysql

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

安装aiomysql

依赖

  • python3.4+
  • asyncio
  • pymysql

安装

  • ?
  • 1
    		•
  • pip install aiomysql
    		•

    应用

    基本的异步连接connection

  • ?
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
    		•
  • 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

  • ?
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
    		•
  • 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

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

  • ?
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
    		•
  • 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))
    		•

    希望本文所述对大家python程序设计有所帮助。

    原文链接：https://blog.csdn.net/ydyang1126/article/details/78226701

    标签：Python mysql 异步操作 aiomysql
  • 上一篇：mysql order by id查询变慢（MySQL不使用order by实现排名的三种思路总结）
  • 下一篇：linuxdocker命令使用教程（seata docker 高可用部署的详细介绍）
    • 您可能感兴趣
    首页 编程学习 Web前端 数据库 软件设计
    开心学习 ©2013-2021 保留所有权利