Python异步操作MySQL完整指南：aiomysql安装、连接池与SQLAlchemy集成

Leo

发布时间：2026-10-13 | 3人浏览过

原创

本文详细介绍如何在Python中通过aiomysql库实现MySQL的异步操作。内容涵盖环境依赖（Python3.4+、asyncio、PyMySQL）、aiomysql的安装方法，以及三种核心应用场景：基础异步连接、异步连接池的使用，以及结合SQLAlchemy进行对象关系映射（ORM）的完整代码示例。

Python异步MySQL操作概述

本文通过具体实例演示如何使用Python进行MySQL的异步操作，供读者参考学习。

aiomysql安装与环境依赖

首先需要安装aiomysql库。

在开始之前，请确保满足以下依赖条件：

Python 3.4或更高版本
asyncio模块
PyMySQL库

执行以下命令进行安装：

aiomysql安装命令


pip install aiomysql

aiomysql核心应用场景

接下来介绍aiomysql在实际应用中的几种常见用法。

基础异步连接示例

首先是建立基本的异步数据库连接。

基础连接代码实现


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

异步连接池使用示例

其次，演示如何创建和管理异步连接池以提高性能。

连接池代码实现


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 ORM集成示例

最后，介绍如何结合SQLAlchemy对象关系映射（ORM）进行数据库操作。

该方法允许用户灵活定义表结构，并轻松执行查询、插入等数据库操作。

SQLAlchemy异步ORM代码实现


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