02-09 Python库-PyMySQL-连接数据库 python测试

需要连接数据库的场景

比如注册接口，同一个账号不能重复注册
比如流程性接口：登陆-绑定银行卡-解绑银行卡测试解绑银行卡的时候，需要先造一个绑定银行卡的数据
修改状态的接口，比如一个接口有几种状态：已绑定-使用中-已注销测试接口不同的状态时候，需还原数据
数据清理，测试完成清理垃圾数据

安装
【02-09 Python库-PyMySQL-连接数据库】pip install PyMySQL
连接数据库

import pymysql# 数据库信息 db_info = { "host": "127.0.0.1",# ip "user": "root",# 用户名 "password": "123456",# 密码 "port": 3306}# 端口class DbConnect(): def __init__(self, db_cof, database=""): self.db_cof = db_cof # 打开数据库连接 self.db = pymysql.connect(database=database, cursorclass=pymysql.cursors.DictCursor, **db_cof)# 使用cursor()方法获取操作游标 self.cursor = self.db.cursor()def select(self, sql): # SQL 查询语句 # sql = "SELECT * FROM EMPLOYEE \ #WHERE INCOME > %s" % (1000) self.cursor.execute(sql) results = self.cursor.fetchall() return resultsdef execute(self, sql): # SQL 删除、提交、修改语句 # sql = "DELETE FROM EMPLOYEE WHERE AGE > %s" % (20) try: # 执行SQL语句 self.cursor.execute(sql) # 提交修改 self.db.commit() except: # 发生错误时回滚 self.db.rollback()def close(self): # 关闭连接 self.db.close()def select_sql(select_sql): '''查询数据库''' db = DbConnect(dbinfo, database="apps") result = db.select(select_sql)# 查询 db.close() return resultdef execute_sql(insert_sql): '''执行sql''' db = DbConnect(dbinfo, database="apps") db.execute(insert_sql)# 查询 db.close()