[PyMySQL]

1. 掌握pymysql对数据库实现增删改查
2. 数据库工具类封装,数据库操作应用场景

数据库操作应用场景

• 校验测试数据 : 删除员工 :
• 构造测试数据 :
  • 测试数据使用一次就失效,不能重复使用 : 添加员工(is_delete)
  • 测试数据在展开测试前无法确定是否存在 : 查询,修改,删除员工

操作步骤:!~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~!

1. 导包 import pymysql
2. 创建链接 : con=pymysql.connect()
3. 创建游标 : cursor=conn.cursor()
4. 执行SQL语句 : conn.execute(“sql语句”)
   • 查询语句selete : 不修改数据库,返回结果集
     • 从结果集中提取想要的数据,cursor.fetch*
   • 增删改insert,update,delete : 没有结果返回,会修改数据库
     • 执行成功 : 提交事务,conn.commit()
     • 执行失败 : 回滚事务,conn.rollback()
5. 关闭游标 : cursor.close()
6. 关闭链接 : conn.close()

事务

• 概念 : 将数据库中连续的一条或多条SQL语句,看作一个整体,作为一个事务存在,要么都成功,要么都失败(eg : 跨行转账)
• 操作方法 :
  • 提交事务 : commit
  • 回滚事务 : rollback

PyMySQL连接数据库

建立连接方法

conn = pymysql.connect(host="",port=0,user="",password="",database="",charset="")
host : 数据库主机ip地址
port : int类型,数据库使用端口号
user : 数据库用户名
password : 数据库用户密码
database : 数据库名
charset : 字符集conn : 成功建立的数据库对象

# 示例----获取版本
# 导包
import pymysql# 创建连接
conn = pymysql.connect(host="  ", port=3306, user="root", password="123456",database="emp", charset='utf8')
# 创建游标
my_cursor=conn.cursor()# 执行select sql
my_cursor.execute("select version();")# 提取结果
res=my_cursor.fetchone()
print("查询结果:",res)
# 关闭游标
my_cursor.close()
# 关闭连接
conn.close()

游标特性

1. 游标负责提取结果集中的数据
2. 新创建的游标指向结果集0条记录
3. 游标提取数据的方式:每次提取游标指向位置的下一条记录
4. 每成功提取一条记录,游标会自动向下游动

提结果集

• 常用方法 :
  • fetchone() : 从结果集中提取一条记录
  • fetchmany(size) : 从结果集中,提取size条记录
  • fetchall() : 提取全部数据
  • 游标属性rownumber : 设置游标位置,常用于归0,cursor.rownumber=0

# 查询t_book表单,获取第一条数据
# 查询t_book表单,获取全部数据
# 查询t_book表,获取第三条和第四条数据
# 导包
import pymysql# 创建连接
conn=pymysql.connect(host="",port=7061,user="root", password="123456",database="emp", charset='utf8')
# 创建游标
cursor1=conn.cursor()
# 执行sql命令
cursor1.execute('select * from t_book')
# 获取1个结果
resp1=cursor1.fetchone(1)
print("提取一个结果:",resp1)# 设置游标归0
cursor1.rownumber=0
# 获取全部结果
resp2=cursor1.fetchall()
print("提取全部:",resp2)# 设置游标指向2
cursor1.rownumber=2# 获取3,4两个结果
resp3=cursor1.fetchmany(2)
print("提取3,4结果:",resp3)# 关闭游标
cursor1.close()
# 关闭连接
conn.close()

异常捕获

# 语法:
try:尝试执行的代码
except Exception as err:出现异常后,执行的代码
finally:无论有没有异常,都会执行的代码
---------------------------------
!!!给cursor和conn创建全局变量,方便全局调用
conn=None
cursor=None

增删改数据

• 使用try-finally抛出异常时,将conn和cursor设为全局变量,
  

# 单独实现如下操作:
1>.新增一条图书数据(id:5 title:西游记 pub_date:1986-01-01)
# 定义两个全局变量
conn=None
my_cursor=None
try:my_cursor.execute("insert into t_book(id,title,pub_date) values(5,'西游记','1986-01-01');")
# 提交事务-------提交事务才会对实际表做修改conn.commit()
如果出现异常,要在except Exception as err中添加回滚事务conn.rollback()
except Exception as err:conn.rollback()

2>.把图书名为'西游记'的阅读量加1
import pymysql# 创建连接
conn=pymysql.connect(host="",port=7061,user="root", password="123456",database="books", charset='utf8',autocommit=False)# 关闭自动提交,改为手动提交
# 创建游标
cursor1=conn.cursor()
# 执行sql命令
# 这里read是关键字名,需要用``确保数值
cursor1.execute("update t_book set `read`=`read`+1 where title = '西游记';")
# 显示影响结果
print("Affected rows:",conn.affected_rows)
# 提交事务
conn.commit()# 关闭游标
cursor1.close()
# 关闭连接
conn.close()

• 执行删除,需要删除的数据如果不存在,不报错,sql语句有问题才报错

3>.删除名称为'西游记'的图书
cursor1.execute("delete t_book where title = '西游记';")
print("Affected_rows",conn.affected_rows())

数据库工具类封装

• 封装的目的 : 将数据库常用的操作,封装成类中的方法,使用时提供要执行的SQL语句给方法即可执行
• 使用者无需担心建立连接,创建游标,关闭游标连接等
• 查询一条记录

# 导包
import pymysql
# 定义数据库工具类
class DBTools(object):"""创建连接-类方法,可以直接使用类名调用"""# 1.类方法实现create,方便将来使用,改self-->cls# 2.谁调用,就返回给谁一个连接# 3.查询记录先创建链接,用类名.方法名调用,变量接收调用返回值# 4.设置为私有方法,外部用户不可见#  def __create_conn(cls)@classmethoddef create_conn(cls):conn1=pymysql.connect(host="",port=7061,user="root", password="123456",database="books", charset='utf8')return conn1@classmethod# 查一条记录def query_one(cls,sql):# 这里设为None是由于都定义在try里面,finally识别不出来.my_conn=Nonemy_cursor=Noneres=Nonetry:"""创建连接,借助类名调用类方法 create_conn"""my_conn=DBTools.create_conn()# 创建游标my_cursor=my_conn.cursor()# 执行语句my_cursor.execute(sql)# 提取一条记录res=my_cursor.fetchone()except Exception as err:print("执行查询SQL失败:",str(err))finally:# 关闭游标my_cursor.close()# 关闭连接my_conn.close()# return 之后不能再执行其它语句,所以关闭写前面return res
# 调用
if __name__ == '__main__':result=DBTools.query_one("select * from emp;")print("查询语句的结果:",result)

• 增删改记录

    # 增删改记录@classmethoddef db_uid(cls,sql):# 创建连接my_conn=DBTools.create_conn()# 创建游标my_cursor=my_conn.cursor()# 执行增删改语句my_cursor.execute(sql)print("Affected rows:",my_conn.affected_rows())my_conn.commit()# 关闭游标my_cursor.close()# 关闭连接my_conn.close()# 调用
if __name__ == '__main__':DBTools.db_uid("update t_book set `read`=100 where id = 3;")

删除员工接口数据库操作

• 构造测试数据(删除员工id)
  • 在测试删除员工前,通过insert语句插入一个员工id到数据库中
  • 直接使用这个id进行删除
• 检验测试数据:
  • 删除员工接口请求成功后,通过select查询is_delete字段的值,为1则删除成功!
  • 使用断言实现
• 基本实现

def test_delete_emp_ok(self):resp=requests.delete(url='http://ihrm-java.itheima.net/api/sys/user/77889900',headers={'Authorization':'61adb484-cdf4-4761-8276-17874caaaf9c'})# 打印响应结果
print(resp.json())

• 构造测试数据

    def setup(self):# 使用直接封装的 数据库工具,执行insert语句,插入一条带有id记录到数据库中del_sql="insert into bs_user(id,mobile,username) values('77889900','13943970710','abc001');"DBTools.db_uid(del_sql)def teardown(self):del_sql="delete from bs_user where id = 77889900;"DBTools.db_uid(del_sql)

• 校验is_delete字段

 # 断言77889900所对应的is_delete记录为1res=DBTools.query_one("select is_delete from bs_user where id=77889900;")assert 1 == res[0]

• 完整代码实现

import pymysql
import requestsfrom py08_db_tools import DBToolsclass TestIhrmDelete(object):def setup(self):# 使用直接封装的 数据库工具,执行insert语句,插入一条带有id记录到数据库中del_sql="insert into bs_user(id,mobile,username) values('77889900','13943970710','abc001');"DBTools.db_uid(del_sql)def teardown(self):del_sql="delete from bs_user where id = 77889900;"DBTools.db_uid(del_sql)def test_delete_emp_ok(self):resp=requests.delete(url='http://ihrm-java.itheima.net/api/sys/user/77889900',headers={'Authorization':'61adb484-cdf4-4761-8276-17874caaaf9c'})# 打印响应结果print(resp.json())assert 200 == resp.status_codeassert True == resp.json().get("success")assert 10000 == resp.json().get("code")assert "操作成功" == resp.json().get("message")# 断言77889900所对应的is_delete记录为1res=DBTools.query_one("select is_delete from bs_user where id=77889900;")assert 1 == res[0]