Python之MySQL数据库连接驱动aiomysql的使用
在上一篇博文介绍了MySQL数据库取得pymysql的使用,参考:
本文介绍异步MySQL异步驱动aiomysql的使用
1,安装异步模块
如果没有模块则先使用pip安装模块
pip3 install asyncio pip3 install aiomysql
2,创建MySQL数据库连接池
和同步方式不一样的是使用异步不能直接创建数据库连接conn,需要先创建一个数据库连接池对象__pool通过这个数据库连接池对象来创建数据库连接
数据库配置信息和介绍pymysql同步使用的数据库是一样的
import asyncio,aiomysql,time # 数据库配置dict db_config = { 'host': 'localhost', 'user': 'www-data', 'password': 'www-data', 'db': 'awesome' } # 创建数据库连接池协程函数 async def create_pool(**kw): global __pool __pool = await aiomysql.create_pool( host=kw.get('host', 'localhost'), port=kw.get('port', 3306), user=kw['user'], password=kw['password'], db=kw['db'] ) loop=asyncio.get_event_loop() loop.run_until_complete(create_pool(**db_config)) # 在事件循环中运行了协程函数则生成了全局变量__pool是一个连接池对象print(__pool) #
3,创建执行sql语句的协程函数
因为是异步模块,只能在事件循环中通过await关键字调用,使用需要创建执行sql语句的协程函数
在协程函数内使用全局上一步创建的连接池对象来创建连接conn和浮标对象cur,通过浮标对象来执行sql语句,执行方法和pymysql模块的执行方法是一样的
cursor.execute(sql,args) sql # 需要执行的sql语句例如'select * from table_name' args # 替换sql语句的格式化字符串,即sql语句可以使用%s代表一个字符串,然后在args中使用对应的变量或参数替换,args为一个list或元组,即是一个有序的序列需要和sql中的%s一一对应 # 例如sql='select * from table_name where id=%s' args=['12345'] # 相当于使用args中的参数替换sql中的%s # select * from table_name where id='12345'
下面分别创建两个协程函数select execute一个用来执行搜索操作,一个用来执行insert,update,delete等修改操作
# 执行select函数 async def select(sql,args,size=None): with await __pool as conn: cur = await conn.cursor(aiomysql.DictCursor) await cur.execute(sql.replace('?','?s'),args or ()) if size: rs = await cur.fetchmany(size) else: rs = await cur.fetchall() await cur.close() return rs # 执行insert update delete函数 async def execute(sql,args): with await __pool as conn: try: cur = await conn.cursor() await cur.execute(sql.replace('?','%s'),args) affetced = cur.rowcount await conn.commit() await cur.close() except BaseException as e: raise return affetced
4,实践执行sql语句
实践执行sql语句前我们首先在本机创建一个数据库和对应的表用于测试
数据库对应的主机,用户名,密码,库名,表名如下
host: localhost user: www-data password: www-data db:awesome table_name: users
创建表名的sql语句如下,需要在数据库中创建好对应的表
CREATE TABLE `users` ( `id` varchar(50) NOT NULL, `email` varchar(50) NOT NULL, `passwd` varchar(50) NOT NULL, `admin` tinyint(1) NOT NULL, `name` varchar(50) NOT NULL, `image` varchar(500) NOT NULL, `created_at` double NOT NULL, PRIMARY KEY (`id`), UNIQUE KEY `idx_email` (`email`), KEY `idx_created_at` (`created_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8
创建好的表对应的结构如下
mysql> desc users; +------------+--------------+------+-----+---------+-------+ | Field | Type | Null | Key | Default | Extra | +------------+--------------+------+-----+---------+-------+ | id | varchar(50) | NO | PRI | NULL | | | email | varchar(50) | NO | UNI | NULL | | | passwd | varchar(50) | NO | | NULL | | | admin | tinyint(1) | NO | | NULL | | | name | varchar(50) | NO | | NULL | | | image | varchar(500) | NO | | NULL | | | created_at | double | NO | MUL | NULL | | +------------+--------------+------+-----+---------+-------+ 7 rows in set (2.68 sec)
①执行insert操作
# insert start import time sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' args = ['test@qq.com','password',1,'test','about:blank',time.time(),'111111'] async def insert(): await execute(sql,args) loop.run_until_complete(insert()) # insert end
执行方式和pymysql没有区别,不同的是需要在事件循环中使用关键字await调用
执行完毕在MySQL中查看插入的数据
mysql> select * from users; +--------+-------------+----------+-------+------+-------------+------------------+ | id | email | passwd | admin | name | image | created_at | +--------+-------------+----------+-------+------+-------------+------------------+ | 111111 | test@qq.com | password | 1 | test | about:blank | 1637738541.48629 | +--------+-------------+----------+-------+------+-------------+------------------+ 1 row in set (0.00 sec)
②执行update操作
直接在loop事件循环中执行execute协程函数也可以
# update start import time sql = 'update `users` set `email`=?, `passwd`=?, `admin`=?, `name`=?, `image`=?, `created_at`=? where `id`=?' args = ['test2@qq.com','password',1,'test2','about:blank',time.time(),'111111'] loop.run_until_complete(execute(sql,args)) # update end
执行以后把email和name都修改了
③执行delete操作
# delete start sql = 'delete from `users` where `id`=?' args = ['111111'] loop.run_until_complete(execute(sql,args)) # delete end
同样根据关键字id指定的值删除了这条数据
④执行selete操作
在执行select操作前我们保证数据库里面至少有一条数据
# select start sql = 'select * from users' args = [] rs = loop.run_until_complete(select(sql,args)) print(rs) # select end
这里直接执行搜索的协程函数select根据函数的定义返回的是所有结果的list,元素是查询结果的字典
输出为
[{'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637739212.74493}]
如果结果有多个则使用list的下标取出
补充
同步模块pymysql和异步模块aiomysql执行速度对比
假如我们需要往数据库插入20000条数据,我们分别使用同步模式和异步模式
首先删除数据库所有测试数据
delete from users;
同步的代码
d:/learn-python3/学习脚本/pymysql/use_pymysql.py
import pymysql db_config = { 'host': 'localhost', 'user': 'www-data', 'password': 'www-data', 'db': 'awesome' } # 创建连接,相当于把字典内的键值对传递 # 相当于执行pymysql.connect(host='localhost',user='www-data',password='www-data',db='awesome') conn = pymysql.connect(**db_config) # 创建游标 cursor = conn.cursor(pymysql.cursors.DictCursor) sql = 'select * from users' args = [] # 执行查询返回结果数量 # 执行查询 rs=cursor.execute(sql,args) # 获取查询结果 # 获取查询的第一条结果,返回一个dict,dict元素是查询对应的键值对 # 如果查询结果有多条则执行一次,游标移动到下一条数据,在执行一次又返回一条数据 # print(cursor.fetchone()) # print(cursor.fetchone()) # print(cursor.fetchall()) # print(cursor.fetchmany()) # {'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637723578.5734} # 获取查询的所有结果,返回一个list,list元素是dict,dict元素是查询对应的键值对 # print(cursor.fetchall()) # [{'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637723578.5734}] # 获取查询的前几条结果,返回一个dict,dict元素是查询对应的键值对 # print(cursor.fetchmany(1)) # [{'id': '111111', 'email': 'test@qq.com', 'passwd': 'password', 'admin': 1, 'name': 'test', 'image': 'about:blank', 'created_at': 1637723578.5734}] # 执行修改操作 import time # # insert start sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' args = ['test1@qq.com','password',1,'test','about:blank',time.time(),'1111121'] # 使用replace 把'?'替换成'%s' # rs = cursor.execute(sql.replace('?','%s'),args) # print(cursor.rowcount) # conn.commit() # print(rs) # insert end # update start # sql = 'update `users` set `email`=?, `passwd`=?, `admin`=?, `name`=?, `image`=?, `created_at`=? where `id`=?' # args = ['test2@qq.com','password',1,'test2','about:blank',time.time(),'111111'] # print(cursor.execute(sql.replace('?','%s'),args)) # conn.commit() # update end # delete start # sql = 'delete from `users` where `id`=?' # args = ['111111'] # print(cursor.execute(sql.replace('?','%s'),args)) # conn.commit() # delete end # 写成函数调用,函数内部使用了数据库连接对象conn # 可以先定义成全局变量global def select(sql,args,size=None): cursor = conn.cursor(pymysql.cursors.DictCursor) cursor.execute(sql.replace('?','%s'),args or ()) if size: rs = cursor.fetchmany(size) else: rs = cursor.fetchall() cursor.close # logging.info('rows returned: %s' % len(rs)) return rs def execute(sql,args): cursor = conn.cursor(pymysql.cursors.DictCursor) try: cursor.execute(sql.replace('?','%s'),args) # rowcount方法把影响函数返回 rs = cursor.rowcount cursor.close() conn.commit() except: raise return rs start_time = time.time() for n in range(20000): sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' email = 'test%s@qq.com' %n args = [email,'password',1,'test','about:blank',time.time(),n] execute(sql,args) end_time = time.time() # 打印开始和结束时间的差 print(end_time - start_time)
我们使用一个循环20000次往数据库插入数据
执行,插入数据比较多需要等待一段时间输出
D:\learn-python3\函数式编程>C:/Python37/python.exe d:/learn-python3/学习脚本/pymysql/use_pymysql.py 77.46903562545776
可以在数据库查询到这20000条数据,而且这个表的字段created_at存储了创建这条数据的时间戳,我们可以看到,id越往后的时间戳越往后,说明数据是同步按顺序一一插入的
我们按照字段created_at排序查询
下面我们删除所有数据使用异步方式插入
异步的代码如下
d:/learn-python3/学习脚本/aiomysql/use_aiomysql.py
import asyncio,aiomysql,time # 数据库配置dict db_config = { 'host': 'localhost', 'user': 'www-data', 'password': 'www-data', 'db': 'awesome' } # 创建数据库连接池协程函数 async def create_pool(**kw): global __pool __pool = await aiomysql.create_pool( host=kw.get('host', 'localhost'), port=kw.get('port', 3306), user=kw['user'], password=kw['password'], db=kw['db'] ) loop=asyncio.get_event_loop() loop.run_until_complete(create_pool(**db_config)) # 在事件循环中运行了协程函数则生成了全局变量__pool是一个连接池对象print(__pool) # # 执行select函数 async def select(sql,args,size=None): with await __pool as conn: cur = await conn.cursor(aiomysql.DictCursor) await cur.execute(sql.replace('?','?s'),args or ()) if size: rs = await cur.fetchmany(size) else: rs = await cur.fetchall() await cur.close() return rs # 执行insert update delete函数 async def execute(sql,args): with await __pool as conn: try: cur = await conn.cursor() await cur.execute(sql.replace('?','%s'),args) affetced = cur.rowcount await conn.commit() await cur.close() except BaseException as e: raise return affetced # insert start # import time # sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' # args = ['test@qq.com','password',1,'test','about:blank',time.time(),'111111'] # async def insert(): # await execute(sql,args) # loop.run_until_complete(insert()) # insert end # update start # import time # sql = 'update `users` set `email`=?, `passwd`=?, `admin`=?, `name`=?, `image`=?, `created_at`=? where `id`=?' # args = ['test2@qq.com','password',1,'test2','about:blank',time.time(),'111111'] # loop.run_until_complete(execute(sql,args)) # update end # delete start # sql = 'delete from `users` where `id`=?' # args = ['111111'] # loop.run_until_complete(execute(sql,args)) # delete end # select start # sql = 'select * from users' # args = [] # rs = loop.run_until_complete(select(sql,args)) # print(rs) # select end async def insert1(): for n in range(10000): sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' email = 'test%s@qq.com' %n args = [email,'password',1,'test','about:blank',time.time(),n] await execute(sql,args) async def insert2(): for n in range(10001,20001): sql = 'insert into `users` (`email`, `passwd`, `admin`, `name`, `image`, `created_at`, `id`) values (?, ?, ?, ?, ?, ?, ?)' email = 'test%s@qq.com' %n args = [email,'password',1,'test','about:blank',time.time(),n] await execute(sql,args) async def main(): # 需要组合成一个事件才会异步执行即在执行insert1的时候同步执行insert2 await asyncio.gather(insert1(),insert2()) start_time = time.time() loop.run_until_complete(main()) end_time = time.time() print(end_time - start_time)
这里我们定义了两个协程函数,分别用来执行前10000个数据和后10000个数据的插入,在main()把这两个协程函数组合成一个事件循环
等待一段时间后执行输出如下,忽略这个warning,可以看到执行时间明显比同步时间短
d:/learn-python3/学习脚本/aiomysql/use_aiomysql.py:42: DeprecationWarning: with await pool as conn deprecated, useasync with pool.acquire() as conn instead with await __pool as conn: 39.794615507125854
我们去数据库查询一下数据也可以看到id从0开始和id从10001开始几乎是同时插入的