每日一模块：aiomysql

一、简介

aiomysql 是一个基于 asyncio 的 MySQL 客户端库，它允许你在 Python 异步编程环境中与 MySQL 数据库进行交互。通过使用 aiomysql，你可以在不阻塞主线程的情况下执行数据库操作，提高应用程序的响应性和吞吐量。

二、安装

你可以使用 pip 来安装 aiomysql：

pip install aiomysql

三、连接到 MySQL 数据库

下面是一个连接到 MySQL 数据库的基本示例：

# database.py  

import aiomysql  
  
# 全局连接池变量  
_pool = None  
  
async def get_pool():  
    global _pool  
    if _pool is None or _pool.closed:  
        _pool = await aiomysql.create_pool(  
            host='localhost',  
            port=3306,  
            user='your_username',  
            password='your_password',  
            db='your_database',  
            autocommit=True  
        )  
    return _pool  
  
async def close_pool():  
    global _pool  
    if _pool is not None and not _pool.closed:  
        _pool.close()  
        await _pool.wait_closed()  
  
# 你可以添加更多的数据库操作函数，比如查询、插入等

四、基本操作

一般我们采用async with 语句来自动管理连接和游标。这意味着当你离开 with 块时，连接和游标会自动关闭。

1. 查询

下面是一个执行 SELECT 查询并打印结果的示例：

import async
from database import get_pool

async def execute_query(pool, sql,args=()):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(sql,args)
            rows = await cur.fetchall()
            for row in rows:
                print(row)

# 示例 SQL 查询
sql = "SELECT * FROM your_table"

# 运行异步函数
asyncio.run(execute_query(await get_pool(), sql))

2. 插入数据

import async
from database import get_pool

async def insert_data(pool, data):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("INSERT INTO your_table (column1, column2) VALUES (%s, %s)", data)

# 示例数据
data = ('value1', 'value2')

# 运行异步函数
asyncio.run(insert_data(await get_pool(), data))

3. 更新数据

import async
from database import get_pool

async def update_data(pool, where_clause, data):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(f"UPDATE your_table SET column1=%s WHERE {where_clause}", data)
            await conn.commit()  # 提交更改

# 示例更新条件和数据
where_clause = "id = 1"
data = ('updated_value',)

# 运行异步函数
asyncio.run(update_data(await get_pool(), where_clause, data))

4. 删除数据

import async
from database import get_pool

async def delete_data(pool, where_clause):
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute(f"DELETE FROM your_table WHERE {where_clause}")
            await conn.commit()  # 提交更改

# 示例删除条件
where_clause = "id = 1"

# 运行异步函数
asyncio.run(delete_data(await get_pool(), where_clause))

五、关闭连接池

通常，连接池会在你的应用程序结束时自动关闭。但是，如果你需要在某个特定时间点关闭连接池，你可以调用连接池的 close() 方法。但是，请注意，如果你打算在应用程序的整个生命周期中重用连接池，那么通常不需要手动关闭它。

async def close_pool(pool):
    pool.close()
    await pool.wait_closed()  # 等待所有池子中的连接都执行完了就关闭

# 运行异步函数关闭连接池（通常不需要这样做，除非你确定不再需要连接池）
# asyncio.run(close_pool(your_pool))

六、事务

在aiomysql中，事务的回滚通常通过执行ROLLBACK SQL语句来实现。当你在一个事务中执行了多个操作，并且希望撤销这些操作（例如，由于某个操作失败或你想撤销之前的更改），你可以使用ROLLBACK语句。

以下是一个如何在aiomysql中使用事务并进行回滚的示例：

import aiomysql
import asyncio

async def perform_transaction(pool, sql_list):
    async with pool.acquire() as conn:
        try:
            async with conn.cursor() as cur:
                # 开始一个事务
                await conn.begin()
                
                # 执行SQL语句列表
                for sql in sql_list:
                    await cur.execute(sql)

                # 如果没有错误，提交事务
                await conn.commit()
        except Exception as e:
            # 如果发生异常，回滚事务
            print(f"An error occurred: {e}")
            await conn.rollback()

# 示例SQL语句列表
sql_list = [
    "UPDATE your_table SET column1='value1' WHERE id=1",
    "UPDATE your_table SET column2='value2' WHERE id=2",  # 假设这行会失败，比如因为约束冲突
    # ... 其他SQL语句
]

# 运行异步函数执行事务
asyncio.run(perform_transaction(await get_pool(), sql_list))

# 注意：这里假设get_pool()函数是从之前的database.py模块中获取的