mysql安装教程 mysql语法大全 python 模块pymysql模块,连接mysql数据库 一、介绍 aioMysql 是一个基于 asyncio 的异步 mysql 客户端库,用于在 pyt
mysql安装教程
mysql语法大全
python 模块pymysql模块,连接mysql数据库
aioMysql 是一个基于 asyncio 的异步 mysql 客户端库,用于在 python 中与 Mysql 数据库进行交互。它提供了异步的数据库连接和查询操作,适用于异步编程环境
pip install aiomysql
import aiomysqlimport asyncioasync def main(): conn = await aiomysql.connect(host='localhost', port=3306, user='root', password='root',db='job') cursor = await conn.cursor() await cursor.execute('SELECT * FROM baidu_job;') result = await cursor.fetchall() print(result) await cursor.close() conn.close()asyncio.run(main())
import aiomysqlimport asyncioasync def main(): pool = await aiomysql.create_pool(host='localhost', port=3306, user='root', password='root',db='job') async with pool.acquire() as conn: async with conn.cursor() as cursor: await cursor.execute('SELECT * FROM baidu_job;') result = await cursor.fetchall() print(result)asyncio.run(main())
'''网址:https://www.che168.com/china/a0_0msdgscncgpi1ltocsp7exf4x0/?pvareaid=102179#currengpostion'''import asyncioimport aioHttpimport aiomysqlfrom lxml import etreeimport randomclass 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())
来源地址:https://blog.csdn.net/randy521520/article/details/132713887
--结束END--
本文标题: python模块之 aiomysql 异步mysql
本文链接: https://www.lsjlt.com/news/418544.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
下载Word文档到电脑,方便收藏和打印~
2024-06-06
2024-06-06
2024-06-05
2024-06-04
2024-06-04
2024-06-03
2024-06-03
2024-06-03
2024-06-04
2024-06-03
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
0