1 import pymysql 2 # 資料庫連線資訊 3 # 多個庫要有多個conn 4 conn = pymysql.connect( 5 host="主機", 6 user="使用者名稱", 7 password="密碼", 8 database="庫名" 9 ) 10 conn1 = pymysql.connect( 11 host="主機", 12 user="使用者名稱", 13 password="密碼", 14 database="庫名" 15 ) 16 17 # 建立遊標物件 18 cursor = conn.cursor() 19 cursor1 = conn1.cursor() 20 21 # 執行對應的SQL 22 cursor.execute 23 # 獲取執行結果 24 Result=cursor.fetchall()
場景一:基於已有的csv檔案,分批次讀取csv檔案中的欄位值作為變數填充到執行的SQL語句
1 csv_file_path = 'csv檔案目錄' 2 with open(csv_file_path, 'r',encoding='utf-8') as file: 3 reader = csv.reader(file) 4 next(reader) # Skip the header row 5 6 batch_size = 100 # 每批次處理的數量 7 total_items = 3100 # 總共需要處理的數量 8 9 for i in range(0, total_items, batch_size): 10 # 在每次迴圈中處理 batch_size 個專案 11 # 可以在迴圈體內部使用 i 作為起始索引 12 13 for j in range(i, min(i + batch_size, total_items)): 14 row = next(reader) 15 # 列印這一行的資料 16 print(row)
場景二:隨機生成特殊欄位的值,作為變數填充到Insert語句中
1 import random 2 import string 3 def generate_credit_code(): 4 # 生成第1位登記管理部門程式碼 5 管理部門程式碼 = ['1', '2', '3', '4', '5', '6', '7', '8', '9', 'A', 'B', 'C', 'D'] 6 register_department = random.choice(管理部門程式碼) 7 # print('管理部門程式碼為',register_department) 8 9 # 生成2-9位組織機構程式碼 10 organizations_code = [] 11 for _ in range(8): 12 org_code = '' 13 for _ in range(8): 14 org_code += random.choice(string.ascii_uppercase + string.digits) 15 organizations_code.append(org_code) 16 organizations_code=random.choice(organizations_code) 17 # print('組織機構程式碼為',organizations_code) 18 19 20 # 生成10-17位統一社會信用程式碼 21 unification_credit_code = '' 22 for _ in range(8): 23 unification_credit_code += random.choice(string.ascii_uppercase + string.digits) 24 # print('統一社會信用程式碼為',unification_credit_code) 25 26 # 組合統一社會信用程式碼 27 credit_code = f"{register_department}{''.join(organizations_code)}{unification_credit_code}" 28 return credit_code
1 mport random 2 3 #這個註冊號是由15個亂數字組成的,使用random.choice方法從0-9中隨機選擇數位。這個方法會被呼叫15次,每次都會生成一個亂數字,然後通過字串的join方法將這15個數位拼接在一起。 4 def generate_reg_code(): 5 # 15位註冊號,以0開頭 6 reg_code = ''.join(random.choice('0123456789') for i in range(15)) 7 return reg_code
(有些亂。。。
-- config.ini 存放的是系統固定的url之類的
-- conftest.py 一般用於放登入介面,使用者返回token,利用fixture被其他介面使用
1 import pytest 2 import requests 3 import pymysql 4 from config import readconfig 5 readcon = readconfig.Read() 6 7 8 @pytest.fixture(scope="session") 9 # 這個方法是pytest封裝公共方法的一個檔案,檔名必須是(conftest.py) 10 # 作用: 其他地方在使用這個方法時就不用from XX import cc 然後也不用範例化了 11 12 13 def test_login(): 14 msg = { 15 "username": '使用者名稱', 16 "password": '加密後的密碼' 17 } 18 19 url =readcon.get_URL("baseurl") 20 cc = requests.post(url+"api/uxxxxxxr/login", params=msg) 21 getjson = cc.json() 22 23 # 獲取token 24 tok = getjson['data']['token'] 25 userid = getjson['data']['userId'] 26 return tok, userid
import pytest import requests from config import readconfig read = readconfig.Read() class TestCase1: global url, tim # 全域性變數,便於其他地方呼叫 url = read.get_URL("baseurl") tim = read.get_URL("timeout") def test_case1(self, test_login): head = {'Content-Type': 'application/json', 'Authorization': test_login[0]} # test_login[0]為token NewtestCreditCodeList = [] SelectNewtestGs = "select 欄位1,欄位2,欄位3 from 資料表 order by id desc limit 100" cursor.execute(SelectNewtestGs) SelectNewtestResult = cursor.fetchall() for tuple in SelectNewtestResult: NewtestCreditCodeList.append(tuple[2]) NewtestGsCreditCodeListResult = ', '.join('"' + i + '"' for i in NewtestCreditCodeList) print('結果為', NewtestGsCreditCodeListResult) r = requests.get(url + 'api/exxxxxh/txxx/xxxxx?入參='+NewtestGsEidListResult, headers=head) print(r.json())