• 微信公众号:美女很有趣。 工作之余,放松一下,关注即送10G+美女照片!

Python之MySQL数据库连接驱动aiomysql的使用

开发技术 开发技术 4小时前 1次浏览

  在上一篇博文介绍了MySQL数据库取得pymysql的使用,参考:https://www.cnblogs.com/minseo/p/15597428.html

  本文介绍异步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是一个连接池对象 <aiomysql.pool.Pool object at 0x00000244AD1724C8>
print(__pool)
# <aiomysql.pool.Pool object at 0x00000244AD1724C8>

  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}
# 获取查询的所有结果,返回一个listlist元素是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排序查询

Python之MySQL数据库连接驱动aiomysql的使用

 

 

  下面我们删除所有数据使用异步方式插入

  异步的代码如下

  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是一个连接池对象 <aiomysql.pool.Pool object at 0x00000244AD1724C8>
print(__pool)
# <aiomysql.pool.Pool object at 0x00000244AD1724C8>

# 执行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开始几乎是同时插入的

Python之MySQL数据库连接驱动aiomysql的使用

 

 

  


程序员灯塔
转载请注明原文链接:Python之MySQL数据库连接驱动aiomysql的使用
喜欢 (0)