Python 之Mysql的异步操作库 aiomysql 笔记

在Python中，进行网络IO操作性能最好的操作就是异步IO，新的Python发布了asyncio库，可以进行异步操作，与 aiohttp类似，python中操作mysql数据库也是网络操作。mysql是c/s体系，每次客户端访问数据库都需要链接mysql服务，发送网络请求，等待mysql数据库服务器返回结果。python中执行异步数据库操作的库为aiomysql。

aiomysql 是一个基于 asyncio 的异步 MySQL 客户端库，用于在 Python 中与 MySQL 数据库进行交互。它提供了异步的数据库连接和查询操作，适用于异步编程环境

1. 异步支持：aiomysql 是基于 asyncio 实现的，可以与 asyncio 框架无缝集成，充分利用异步编程的优势，提高应用程序的性能和并发能力。
2. 高性能：aiomysql 使用了底层的 PyMySQL 库，通过异步操作和连接池技术，可以实现高性能的数据库访问。
3. 支持事务：aiomysql 提供了事务管理的支持，可以执行原子性的数据库操作，保证数据的一致性。
4. SQL 语句构建器：aiomysql 提供了一个 SQL 语句构建器，可以方便地构建和执行 SQL 查询。

一、安装

pip install aiomysql

二、aiomysql.connect()

在操作mysql之前，需要与服务器进行链接。connect方法用于与mysql服务器链接，链接成功后返回一个链接对象，用于执行后续操作。

使用示例

import aiomysql
import asyncio
 
 
async def main():
    conn = await aiomysql.connect(host='localhost', port=3306, user='root', password='123456',db='job')
    cursor = await conn.cursor()
    await cursor.execute('SELECT * FROM hxcms_blog;')
    result = await cursor.fetchall()
    print(result)
    await cursor.close()
    conn.close()
 
 
asyncio.run(main())

示例2：

 

import asyncio
import aiomysql
 
#定义操作协程
async def main():
    # 连接Mysql数据库
    conn = await aiomysql.connect(
    host='localhost',  # Mysql服务器IP地址
    port=3306,  # Mysql服务器端口号
    user='root',  # Mysql用户名
    password='123456',  # Mysql密码
    db='mydb',  # Mysql数据库名称
    autocommit=True)  # 自动提交
    # 配置游标
    cur = await conn.cursor()
    # 执行SQL语句
    await cur.execute('SELECT * FROM `user`')
    # 获取查询结果
    result = await cur.fetchall()
    print(result)
    # 关闭连接
    await cur.close()
    conn.close()
# 启动事件循环，事件循环中将执行 main 协程
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

二、aiomysql.create_pool()

创建一个连接池。该方法用于创建一个异步连接池，接受一系列的连接参数，例如数据库主机、端口、用户名、密码等。

使用示例 

import aiomysql
import asyncio
 
async def main():
    pool = await aiomysql.create_pool(host='localhost', port=3306, user='root', password='123456',db='job')
    async with pool.acquire() as conn:
        async with conn.cursor() as cursor:
            await cursor.execute('SELECT * FROM hxcms_blog;')
            result = await cursor.fetchall()
            print(result)
 
asyncio.run(main())

aiomysql内建对象说明

1、pool-链接池对象

2、connection 数据库连接对象

3、cursor-游标对象

一个综合例子

该例子用于抓取某网站数据。

'''
网址：https://www.che168.com/china/a0_0msdgscncgpi1ltocsp7exf4x0/?pvareaid=102179#currengpostion
'''
 
import asyncio
import aiohttp
import aiomysql
from lxml import etree
import random
 
 
class Car:
    url = {
        'car_list': 'https://www.che168.com/china/a0_0msdgscncgpi1ltocsp{}exf4x0/?pvareaid=102179#currengpostion',
        'car_detail': 'https://cacheapigo.che168.com/CarProduct/GetParam.ashx?specid={}'
    }
 
    headers = {
        'authority': 'www.che168.com',
        'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
        'accept-language': 'zh-CN,zh;q=0.9',
        'cache-control': 'no-cache',
        'pragma': 'no-cache',
        'referer': 'https://www.che168.com/dealer/481320/48787661.html?pvareaid=100519&userpid=0&usercid=0&offertype=&offertag=0&activitycartype=0&fromsxmlist=0',
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'document',
        'sec-fetch-mode': 'navigate',
        'sec-fetch-site': 'same-origin',
        'sec-fetch-user': '?1',
        'upgrade-insecure-requests': '1',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36',
    }
 
    def __init__(self):
        self.pool = None
        self.request = None
 
    '''获取汽车列表'''
    async def get_car_list(self, page):
        print(self.url['car_list'].format(page))
        response = await self.request.get(self.url['car_list'].format(page))
        result = await response.text(encoding='GBK')
        return etree.HTML(result)
 
    '''处理汽车列表信息'''
    async def parse_car_list_info(self, html):
        car_list_html = html.xpath('//div[@id="goodStartSolrQuotePriceCore0"]//li[@name="lazyloadcpc"]')
        task_list = []
        for item in car_list_html:
            car_info = {
                'specid': item.xpath('./@specid')[0],
                'infoid': item.xpath('./@infoid')[0]
            }
            task = asyncio.create_task(self.get_car_detail(car_info))
            task_list.append(task)
        await asyncio.wait(task_list)
 
    '''获取第一页数据，并返回总页数'''
    async def get_first_page(self):
        html = await self.get_car_list(1)
        page_total = html.xpath('//div[@id="listpagination"]/a[last()-1]/text()')[0]
        await self.parse_car_list_info(html)
        return int(page_total)
 
    '''获取除第一页数据之外的其他页数据'''
    async def get_all_page(self,page):
        async with self.semaphore:
            await asyncio.sleep(random.randint(500, 800) / 1000)
            html = await self.get_car_list(page)
            await self.parse_car_list_info(html)
 
    '''获取汽车详情'''
    async def get_car_detail(self, car_info):
        response = await self.request.get(self.url['car_detail'].format(car_info['specid']))
        result = await response.json()
        detail = result['result']['paramtypeitems']
        car_detail = {
            'specid': car_info['specid'],
            'infoid': car_info['infoid'],
            'name': detail[0]['paramitems'][0]['value'],
            'price': detail[0]['paramitems'][1]['value'],
            'manufacturer': detail[0]['paramitems'][2]['value'],
            'level': detail[0]['paramitems'][3]['value'],
            'length': f'{detail[1]["paramitems"][0]["value"]}mm',
            'width': f'{detail[1]["paramitems"][1]["value"]}mm',
            'height':f'{detail[1]["paramitems"][2]["value"]}mm',
        }
        await self.insert_table(car_detail)
 
    '''异步建立mysql表'''
    async def create_table(self):
        sql = '''
        CREATE TABLE IF NOT EXISTS qichezhijia(
        Id INT UNIQUE,
        Specid INT,
        Name VARCHAR(255),
        Price VARCHAR(10),
        Manufacturer VARCHAR(255),
        Level VARCHAR(50),
        Length VARCHAR(10),
        Width VARCHAR(10),
        Height VARCHAR(10),
        PRIMARY KEY(Id)
        )
        '''
        async with self.pool.acquire() as conn:
            async with conn.cursor() as cursor:
                try:
                    await cursor.execute(sql)
                except Exception as e:
                    print(f'创建表失败{e}')
 
    '''插入数据'''
    async def insert_table(self,car_detail):
        sql = '''
        INSERT INTO qichezhijia VALUES(%(infoid)s,%(specid)s,%(name)s,%(price)s,%(manufacturer)s,%(level)s,%(length)s,%(width)s,%(height)s)
        '''
        async with self.pool.acquire() as conn:
            async with conn.cursor() as cursor:
                try:
                    await cursor.execute(sql, car_detail)
                    await conn.commit()
                    print('数据插入成功')
                except Exception as e:
                    print(f'插入数据失败{e}，infoid={car_detail["infoid"]}')
 
    '''程序运行主函数'''
    async def main(self):
        async with aiomysql.create_pool(host='localhost', port=3306, user='root', password='root', db='car') as pool:
            self.pool = pool
            await self.create_table()
            async with aiohttp.ClientSession(headers=self.headers) as request:
                self.request = request
                page_total = await self.get_first_page()
                self.semaphore = asyncio.Semaphore(3)
                task_list = []
                for page in range(2,page_total+1):
                    task = asyncio.create_task(self.get_all_page(page))
                    task_list.append(task)
                await asyncio.wait(task_list)
 
 
if __name__ == '__main__':
    car = Car()
    asyncio.run(car.main())