python 操做數據庫 轉載：https://www.jianshu.com/p/736ed1e724f0

課程目錄

Python接口測試實戰1（上）- 接口測試理論
Python接口測試實戰1（下）- 接口測試工具的使用
Python接口測試實戰2 - 使用Python發送請求
Python接口測試實戰3（上）- Python操做數據庫
Python接口測試實戰3（下）- unittest測試框架
Python接口測試實戰4（上） - 接口測試框架實戰
Python接口測試實戰4（下） - 框架完善：用例基類，用例標籤，從新運行上次失敗用例
Python接口測試實戰5（上） - Git及Jenkins持續集成
Python接口測試實戰5（下） - RESTful、Web Service及Mock Server

更多學習資料請加添加做者微信：lockingfree獲取

本節內容

• 數據庫操做
• 封裝數據庫操做

前言

在功能、接口測試中經常須要經過數據庫的操做，來準備數據、檢測環境及覈對功能、接口的數據庫操做是否正確。
在自動化測試中，就須要咱們用代碼鏈接數據庫自動完成數據準備、環境檢查及數據庫斷言的功能。
使用Python操做MySQL數據庫這裏咱們須要用到三方庫PyMySQl

安裝方法：pip install pymysql

數據庫操做

1. 創建數據庫鏈接 conn = pymysql.connect()
2. 從鏈接創建操做遊標 cur = conn.cursor()
3. 使用遊標執行sql（讀/寫） cur.execute(sql)
4. 獲取結果（讀）/ 提交更改（寫） cur.fetchall() / conn.commit()
5. 關閉遊標及鏈接 cur.close();conn.close()

import pymysql # 1. 創建鏈接 conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123456', # password也能夠 db='api_test', charset='utf8') # 若是查詢有中文須要指定數據庫編碼 # 2. 從鏈接創建遊標（有了遊標才能操做數據庫） cur = conn.cursor() # 3. 查詢數據庫（讀） cur.execute("select * from user where name='張三'") # 4. 獲取查詢結果 result = cur.fetchall() print(result) # 3. 更改數據庫（寫） cur.execute("delete from user where name='李四'") # 4. 提交更改 conn.commit() # 注意是用的conn不是cur # 5. 關閉遊標及鏈接 cur.close() conn.close() 

什麼是遊標？ 遊標相似文件句柄，能夠逐條的訪問數據庫執行結果集。pymysql中只能經過遊標來執行sql和獲取結果

查詢操做
使用cur.execute(), 執行數據庫查詢後無返回的是影響的行數，而非查詢結果。咱們要使用cur.fetchone()/cur.fetchmany()/cur.fetchall()來獲取查詢結果

• cur.fetchone(): 獲取一條數據（同時獲取的數據會從結果集刪除），返回元祖('張三','123456')
• cur.fetchmany(3): 獲取多條數據，返回嵌套元祖(('張三','123456'),('李四','123456'),("王五","123456"))
• cur.fetchall(): 獲取全部數據，返回嵌套元祖，(('張三','123456'),)（只有一條數據時）

注意： 獲取完數據後，數據會從數據集中刪除，再次獲取獲取不到，如：

cur.execute(select * from user where name='張三')
print(cur.fetchone()) # 結果： ('張三','123456')
print(cur.fetchone()) # 結果：None
print(cur.fetchall()) # 結果：()

因此咱們須要重複使用查詢結果時，須要將查詢結果賦給某個變量

cur.execute(select * from user where name='張三') result = cur.fetchall() print(result) # 結果： ('張三','123456') print(result) # 結果： ('張三','123456') 

修改操做
執行修改數據庫的操做後不當即生效，使用鏈接conn.commit()提交後才生效，支持事物及回滾

try: cur.execute("insert into user (name,password) values ('張三', '123456')") cur.execute("insert into user (name, passwd) values ('李四'), '123456'") # 此處sql出錯 conn.commit() # 使用鏈接提交全部更改 except Exception as e: conn.rollback() # 回滾全部更改（注意用的是conn) print(str(e)) 

封裝數據庫操做

因爲常常要使用到數據庫操做，建議將全部數據庫操做封裝成公用的數據庫模塊

1. 新建db.py, 代碼以下：

import pymysql # 獲取鏈接方法 import pymysql # 獲取鏈接方法 def get_db_conn(): conn = pymysql.connect(host='127.0.0.1', port=3306, user='test', passwd='123456', db='api_test', charset='utf8') # 若是查詢有中文，須要指定測試集編碼 return conn # 封裝數據庫查詢操做 def query_db(sql): conn = get_db_conn() # 獲取鏈接 cur = conn.cursor() # 創建遊標 cur.execute(sql) # 執行sql result = cur.fetchall() # 獲取全部查詢結果 cur.close() # 關閉遊標 conn.close() # 關閉鏈接 return result # 返回結果 # 封裝更改數據庫操做 def change_db(sql): conn = get_db_conn() # 獲取鏈接 cur = conn.cursor() # 創建遊標 try: cur.execute(sql) # 執行sql conn.commit() # 提交更改 except Exception as e: conn.rollback() # 回滾 finally: cur.close() # 關閉遊標 conn.close() # 關閉鏈接 # 封裝經常使用數據庫操做 def check_user(name): # 注意sql中''號嵌套的問題 sql = "select * from user where name = '{}'".format(name) result = query_db(sql) return True if result else False def add_user(name, password): sql = "insert into user (name, passwd) values ('{}','{}')".format(name, password) change_db(sql) def del_user(name): sql = "delete from user where name='{}'".format(name) change_db(sql) 

相比用例中直接使用sql操做數據庫，封裝經常使用的數據庫操做會更安全

1. 調用方法(其餘模塊)

from db import * if check_user("張三"): del_user("張三") 

補充：另外一種封裝方法

因爲上面這種封裝方法，每一次查詢都會創建一次數據庫鏈接，效率較低，也能夠採用下面面向對象的封裝方法

db2.py

import pymysql


class DB: def __init__(self): self.conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123456', # passwd 不是 password db='api_test') self.cur = self.conn.cursor() def __del__(self): # 析構函數，實例刪除時觸發 self.cur.close() self.conn.close() def query(self, sql): self.cur.execute(sql) return self.cur.fetchall() def exec(self, sql): try: self.cur.execute(sql) self.conn.commit() except Exception as e: self.conn.rollback() print(str(e)) def check_user(self,name): result = self.query("select * from user where name='{}'".format(name)) return True if result else False def del_user(self, name) self.exec("delete from user where name='{}'".format(name)) 

使用方法

from db2 import DB: db = DB() # 實例化一個數據庫操做對象 if db.check_user("張三"): db.del_user("張三") 

後言

• 數據庫鏈接信息建議寫到配置文件中，從配置文件中讀取
• sql語句建議先在手工測試一下沒有語法問題再進行封裝
• 經過封裝各類sql能夠完成各類業務操做
• 更改數據庫有風險，操做需謹慎！！！

做者：韓志超
連接： https://www.jianshu.com/p/736ed1e724f0 來源：簡書 著做權歸做者全部。商業轉載請聯繫做者得到受權，非商業轉載請註明出處。