歡迎關注【無量測試之道】公眾號，回覆【領取資源】,
Python程式設計學習資源乾貨、
Python+Appium框架APP的UI自動化、
Python+Selenium框架Web的UI自動化、
Python+Unittest框架API自動化、

資源和程式碼 免費送啦~
文章下方有公眾號二維碼，可直接微信掃一掃關注即可。

本文主要是分享如何將db資料刷入到ES中，步驟非常的簡單。
分為兩步：
第一步是連線資料庫，獲取到要匹配的資料。
第二步是呼叫對應的寫ES的介面。

匯入第三方庫

 1 #匯入第三方庫
 2 import pymysql
 3 import requests
 4  
 5 #資料配置定義的一個字典
 

 6 ars3_dicts={
 7     "HOST" : 'l-test.beta',
 8     "PORT" : 3306,
 9     "USER": 'test',
10     "PASSWORD" : 'test123',
11     "NAME":"entity_0"
12 }

執行sql方法

 1 #執行sql方法
 2 def executesql(query):
 3     try:
 4         conn = pymysql.connect(ars3_dicts['HOST'], ars3_dicts['USER'], ars3_dicts['PASSWORD
 
'], ars3_dicts['NAME'], int(ars3_dicts['PORT']),charset='utf8')
 5         cursor = conn.cursor()
 6         cursor.execute(query)
 7         result =cursor.fetchall()
 8         print(result)
 9         print("execute successfully!!!")
10         return list(result)
11     except Exception as e:
12         print
 
(e)
13         print("execute failed")
14     finally:
15         cursor.close()
16         conn.close()

執行寫ES的方法

 1 #執行寫ES的方法
 2 def insertES():
 3  sql="SELECT entity_id FROM entity_1.`entity_main_0`;" #從資料庫中提供要的資料
 4  alist=executesql(sql)
 5  for i in range(len(alist)):
 6      id=alist[i][0]
 7      if(str(id).startswith("3")):
 8          url="http://10.7.36.35:9092/api/profile/backdoor/refreshEsData?arsProfileIds="+str(id) #此處是由開發提供的介面
 9          print(url)
10          response=requests.get(url)
11          print(response.text)

入口方法

1 #入口方法
2 if __name__ == '__main__':
3     inseertES()

請求後列印的資訊如下圖所示：

總結：當測試環境的db被另一個環境的db覆蓋時，所對應的ES 裡面的資料也是要同步進行覆蓋的，本文是通過查詢db中的資料，然後呼叫開發開放的介面進行的初始化，這樣就可以保障頁面檢索的資料與db 是一致的。

備註：我的個人公眾號已正式開通，致力於測試技術的分享，包含：大資料測試、功能測試，測試開發，API介面自動化、測試運維、UI自動化測試等，微信搜尋公眾號：“無量測試之道”，或掃描下方二維碼：

新增關注，讓我們一起共同成長！