Best Python code snippet using autotest_python
mySqlHelper.py
Source:mySqlHelper.py
1import threading2import time3import traceback4import pymysql5from conf import config6# def getFileTemp():7# try:8# conn = pymysql.connect(host=config.dbHost,9# port=config.dbPort,10# user=config.dbUser,11# passwd=config.dbPasswd,12# db=config.dbPlatform,13# charset='utf8')14# except:15#16# log.exception('æ°æ®åºè¿æ¥å¼å¸¸')17# return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '18#19# try:20# # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor21# cursor = conn.cursor()22# cursor = conn.cursor(pymysql.cursors.DictCursor)23# # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢24# sql = 'SELECT * FROM file_temp f WHERE id>3128641 AND (f.type=1 OR f.type=2) ORDER BY id ASC LIMIT 1000'25# sql = 'SELECT * FROM file_temp f WHERE id>3130600 AND f.type=2 and hh_flag=2 ORDER BY id ASC LIMIT 750'26# # sql='SELECT * FROM file_temp f WHERE id>3130600 AND f.type=2 AND hh_flag=2 AND hh_ret LIKE \'%{"result":"1","name":"invoiceMethod","type":0}%\' ORDER BY id ASC LIMIT 100'27# cursor.execute(sql)28#29# data = []30# for row in cursor.fetchall():31# path = row["name"]32# hhFlag = row["hh_flag"]33# uid = row["uid"]34# type = row["type"]35# fileId = row["file_id"]36# config.domain37# # print("hhFlag:", hhFlag, " path:", config.domain + path)38# model = fileTemp(uid, fileId, path, type, hhFlag)39# data.append(model)40# return 0, data, None41# except:42# log.exception('æ°æ®åºæ¥è¯¢å¼å¸¸')43# return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'44# finally:45# # å
³éæ°æ®åºè¿æ¥46# conn.close()47from module.MigCompany import MigCompany48from module.Template import Template49from tools import log50def getTemplateSubjectById(id):51 try:52 conn = pymysql.connect(host=config.dbHost,53 port=config.dbPort,54 user=config.dbUser,55 passwd=config.dbPasswd,56 db=config.dbPlatform,57 charset='utf8')58 except:59 log.exception('æ°æ®åºè¿æ¥å¼å¸¸')60 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '61 try:62 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor63 cursor = conn.cursor()64 cursor = conn.cursor(pymysql.cursors.DictCursor)65 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢66 sql = 'SELECT t.`ID` tId,t.`NAME` tName,t.`SUBJECT_TYPE` subjectType, ts.`ID` tsId,ts.`NAME` tsName,ts.`SUBJECT_ID` kmCode,ts.`SUBJECT_NAME` kmName,ts.`TYPE` jdType ' \67 'FROM `template` t,template_subject ts ' \68 'WHERE ts.`DELETE_FLAG`=1 AND t.`DELETE_FLAG`=1 AND ts.`TEMPLATE_ID`=t.`ID` AND t.`ID`=' + str(id)69 cursor.execute(sql)70 data = []71 for row in cursor.fetchall():72 tId = row["tId"]73 tName = row["tName"]74 tsId = row["tsId"]75 subjectType = row["subjectType"]76 tsName = row["tsName"]77 kmCode = row["kmCode"]78 kmName = row["kmName"]79 jdType = row["jdType"]80 config.domain81 model = Template(tId, tName, subjectType, tsId, tsName, kmCode, kmName, jdType)82 data.append(model)83 return 0, data, None84 except:85 log.exception('æ°æ®åºæ¥è¯¢å¼å¸¸')86 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'87 finally:88 # å
³éæ°æ®åºè¿æ¥89 conn.close()90# def getInfoMapping():91# try:92# conn = pymysql.connect(host=config.dbHost,93# port=config.dbPort,94# user=config.dbUser,95# passwd=config.dbPasswd,96# db=config.dbPlatform,97# charset='utf8')98# except:99#100# log.exception('æ°æ®åºè¿æ¥å¼å¸¸:' )101#102# return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '103#104# try:105# # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor106# cursor = conn.cursor()107# cursor = conn.cursor(pymysql.cursors.DictCursor)108# # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢109# sql = 'SELECT * FROM `info_template_mapping` itm WHERE itm.`org_id`=6 OR itm.`org_id`=1 AND itm.`ID` >668 ORDER BY itm.`ID`'110# cursor.execute(sql)111#112# data = []113# for row in cursor.fetchall():114# id = row["id"]115# template_id = row["template_id"]116# mapping = row["mapping"]117#118# model = infoTemplateMapping(id, template_id, mapping)119# data.append(model)120# return 0, data, None121# except:122# log.exception('æ°æ®åºæ¥è¯¢å¼å¸¸' )123# return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'124# finally:125# # å
³éæ°æ®åºè¿æ¥126# conn.close()127# def getSubjectByTemplateId(id):128# try:129# conn = pymysql.connect(host=config.dbHost,130# port=config.dbPort,131# user=config.dbUser,132# passwd=config.dbPasswd,133# db=config.dbPlatform,134# charset='utf8')135# except:136#137# print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)138# traceback.print_exc()139# return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '140#141# try:142# # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor143# cursor = conn.cursor()144# cursor = conn.cursor(pymysql.cursors.DictCursor)145# # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢146# sql = 'SELECT * FROM `template_subject` ts WHERE ts.`DELETE_FLAG`=1 AND ts.`TEMPLATE_ID`=' + str(id)147# cursor.execute(sql)148#149# data = []150# for row in cursor.fetchall():151# id = row["ID"]152# SUBJECT_CODE = row["SUBJECT_CODE"]153#154# model = templateSubject(id, SUBJECT_CODE)155# data.append(model)156# return 0, data, None157# except:158# traceback.print_exc()159#160# print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)161# return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'162# finally:163# # å
³éæ°æ®åºè¿æ¥164# conn.close()165# def updataNewTemplateMapping(id, mapping):166# try:167# conn = pymysql.connect(host=config.dbHost,168# port=config.dbPort,169# user=config.dbUser,170# passwd=config.dbPasswd,171# db=config.dbPlatform,172# charset='utf8')173# except:174#175# print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)176# traceback.print_exc()177# return -1, 'æ°æ®åºè¿æ¥å¼å¸¸ '178#179# try:180# # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor181# cursor = conn.cursor()182#183# # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢184# sql = "UPDATE `info_template_mapping` SET mapping='" + mapping + "' WHERE id=" + str(id)185# cursor.execute(sql)186# # æ交å°æ°æ®åºæ§è¡187# conn.commit()188#189# return 0, None190# except:191# conn.rollback()192# traceback.print_exc()193#194# print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)195# return -2, 'æ°æ®åºæ¥è¯¢å¼å¸¸'196# finally:197# # å
³éæ°æ®åºè¿æ¥198# conn.close()199# # åç°æ件被å½å
¥æ¶æ´æ°ä¸´æ¶è¡¨ç¶æ 主åº200# #201# def updataFileTempWhenInputed(uid):202# try:203# conn = pymysql.connect(host=config.dbHost, port=config.dbPort, user=config.dbUser,204# passwd=config.dbPasswd, db=config.dbPlatform, charset='utf8')205# except:206# traceback.print_exc()207# return -1208#209# try:210# # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor211# cursor = conn.cursor()212# # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢213# sql = "UPDATE file_management f,file_temp t SET t.status=5 WHERE f.id=t.file_id AND t.uid=%s AND f.file_status=2"214# cursor.execute(sql, uid)215#216# except:217# traceback.print_exc()218# return -2219# finally:220# # å
³éæ°æ®åºè¿æ¥221# conn.close()222def getThirdLog(id):223 try:224 conn = pymysql.connect(host=config.dbHost,225 port=config.dbPort,226 user=config.dbUser,227 passwd=config.dbPasswd,228 db=config.dbLog,229 charset='utf8')230 except:231 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)232 traceback.print_exc()233 return -1, None, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '234 try:235 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor236 cursor = conn.cursor(pymysql.cursors.DictCursor)237 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢238 sql = "SELECT * FROM cs_third_log l WHERE l.id=" + str(id)239 cursor.execute(sql)240 for row in cursor.fetchall():241 # print('æ°æ®åºæ¥:',row["id"])242 # print('æ°æ®åºæ¥:',row["af"])243 # print('æ°æ®åºæ¥:',row["bf"])244 af = row["af"]245 bf = row["bf"]246 return 0, str(af, encoding="utf8"), str(bf, encoding="utf8"), None247 except:248 traceback.print_exc()249 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)250 return -2, None, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'251 finally:252 # å
³éæ°æ®åºè¿æ¥253 conn.close()254def updateTest():255 try:256 conn = pymysql.connect(host="192.168.1.10",257 port=3307,258 user="root",259 passwd="d0608",260 db="csdb20170913153728181",261 charset='utf8')262 except:263 traceback.print_exc()264 return265 try:266 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor267 cursor = conn.cursor()268 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢269 sql = "UPDATE file_management set original_info ='567567567',status = 3 WHERE ID = 5"270 intRet = cursor.execute(sql)271 conn.commit()272 print("intRet", intRet)273 except:274 traceback.print_exc()275 return276 finally:277 # å
³éæ°æ®åºè¿æ¥278 conn.close()279# return returnCode280# max_connections281# processCount282# msg283def getConnStatus(connConfig):284 try:285 conn = pymysql.connect(host=connConfig.dbHost,286 port=connConfig.dbPort,287 user=connConfig.dbUser,288 passwd=connConfig.dbPasswd,289 db=connConfig.dbPlatform,290 charset='utf8')291 except BaseException as e:292 log.e('æ°æ®åºè¿æ¥å¼å¸¸:', e, traceback._context_message)293 traceback.print_exc()294 return -1, None, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '295 try:296 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor297 cursor = conn.cursor(pymysql.cursors.DictCursor)298 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢299 sql = "SHOW VARIABLES LIKE '%max_connections%'"300 cursor.execute(sql)301 for row in cursor.fetchall():302 if row["Variable_name"] == 'max_connections':303 max_connections = row["Value"]304 break305 sql = "SHOW PROCESSLIST"306 cursor.execute(sql)307 processCount = 0308 for row in cursor.fetchall():309 processCount = processCount + 1310 return 0, int(max_connections), processCount, None311 except:312 traceback.print_exc()313 log.e('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)314 return -2, None, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'315 finally:316 # å
³éæ°æ®åºè¿æ¥317 conn.close()318def getUnPdfCount():319 try:320 conn = pymysql.connect(host=config.dbHost,321 port=config.dbPort,322 user=config.dbUser,323 passwd=config.dbPasswd,324 db=config.dbPlatform,325 charset='utf8')326 except:327 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)328 traceback.print_exc()329 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '330 try:331 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor332 cursor = conn.cursor(pymysql.cursors.DictCursor)333 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢334 sql = "SELECT COUNT(*) count FROM `monthly_temp`"335 cursor.execute(sql)336 for row in cursor.fetchall():337 count = row["count"]338 return 0, count, None339 except:340 traceback.print_exc()341 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)342 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'343 finally:344 # å
³éæ°æ®åºè¿æ¥345 conn.close()346def getUnHeHeCount():347 try:348 conn = pymysql.connect(host=config.dbHost,349 port=config.dbPort,350 user=config.dbUser,351 passwd=config.dbPasswd,352 db=config.dbPlatform,353 charset='utf8')354 except:355 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)356 traceback.print_exc()357 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '358 try:359 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor360 cursor = conn.cursor(pymysql.cursors.DictCursor)361 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢362 sql = "SELECT COUNT(*) count FROM `file_temp` f WHERE (f.type=1 OR f.type=2 ) AND f.`hh_flag`=1"363 cursor.execute(sql)364 count = 0365 for row in cursor.fetchall():366 count = row["count"]367 return 0, count, None368 except:369 traceback.print_exc()370 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)371 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'372 finally:373 # å
³éæ°æ®åºè¿æ¥374 conn.close()375def getUnHeHeWaitTime():376 try:377 conn = pymysql.connect(host=config.dbHost,378 port=config.dbPort,379 user=config.dbUser,380 passwd=config.dbPasswd,381 db=config.dbPlatform,382 charset='utf8')383 except:384 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)385 traceback.print_exc()386 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '387 try:388 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor389 cursor = conn.cursor(pymysql.cursors.DictCursor)390 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢391 sql = "SELECT TIMESTAMPDIFF(MINUTE, f.`time`,NOW()) dt, f.* FROM `file_temp` f WHERE (f.type = 1 OR f.type = 2 ) AND f.`hh_flag` = 1 ORDER BY id LIMIT 1"392 cursor.execute(sql)393 minutes = 0394 for row in cursor.fetchall():395 minutes = row["dt"]396 return 0, minutes, None397 except:398 traceback.print_exc()399 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)400 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'401 finally:402 # å
³éæ°æ®åºè¿æ¥403 conn.close()404# çæåè¯çå¾
405def getUnCreateWaitTime():406 try:407 conn = pymysql.connect(host=config.dbHost,408 port=config.dbPort,409 user=config.dbUser,410 passwd=config.dbPasswd,411 db=config.dbPlatform,412 charset='utf8')413 except:414 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)415 traceback.print_exc()416 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '417 try:418 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor419 cursor = conn.cursor(pymysql.cursors.DictCursor)420 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢421 sql = "SELECT TIMESTAMPDIFF(MINUTE, f.`time`,NOW()) dt, f.* FROM `file_temp` f " \422 "WHERE (f.type = 1 OR f.type = 2) AND f.`hh_flag` = 2 AND f.`auto_flg` IS NULL AND " \423 "f.`uid` IN (SELECT uid FROM `fun_ext` k WHERE k. TYPE=1 AND k.`fun`=3) ORDER BY id LIMIT 1 ;"424 cursor.execute(sql)425 minutes = 0426 for row in cursor.fetchall():427 minutes = row["dt"]428 return 0, minutes, None429 except:430 traceback.print_exc()431 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)432 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'433 finally:434 # å
³éæ°æ®åºè¿æ¥435 conn.close() # çæåè¯çå¾
436def insertBalance(kemuyueb, company):437 try:438 conn = pymysql.connect(host='localhost',439 port=3306,440 user='root',441 passwd='root',442 db='lxhwdata',443 charset='utf8')444 except:445 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)446 traceback.print_exc()447 return -1, 'æ°æ®åºè¿æ¥å¼å¸¸ '448 try:449 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor450 cursor = conn.cursor(pymysql.cursors.DictCursor)451 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢452 sql = "INSERT INTO mig_balance (COMPANY_ID,CODE, NAME,OPENING_DEBIT_VALUE,OPENING_CREDIT_VALUE,CURRENT_DEBIT_VALUE," \453 "CURRENT_CREDIT_VALUE,CLOSING_DEBIT_VALUE,CLOSING_CREDIT_VALUE,TYPE) VALUES (%s,%s, %s, %s, %s, %s, %s, %s, %s,2)"454 val = (company, kemuyueb.accountCode, kemuyueb.accountName, kemuyueb.beginningBalanceDebit,455 kemuyueb.beginningBalanceCrebit,456 kemuyueb.currentAmountDebit, kemuyueb.currentAmountCrebit, kemuyueb.endingBalanceDebit,457 kemuyueb.endingBalanceCrebit)458 cursor.execute(sql, val)459 conn.commit()460 return 0, cursor.rowcount461 except:462 traceback.print_exc()463 print('ä¿åä½é¢è¡¨å¼å¸¸:', traceback._context_message)464 return -2, 'æ°æ®åºæ¥è¯¢å¼å¸¸'465 finally:466 # å
³éæ°æ®åºè¿æ¥467 conn.close()468def insertBalanceBaseCode(kemuyueb, company):469 try:470 conn = pymysql.connect(host='localhost',471 port=3306,472 user='root',473 passwd='root',474 db='hycaitestdata',475 charset='utf8')476 except:477 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)478 traceback.print_exc()479 return -1, 'æ°æ®åºè¿æ¥å¼å¸¸ '480 try:481 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor482 cursor = conn.cursor(pymysql.cursors.DictCursor)483 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢484 sql = "INSERT INTO mig_balance (COMPANY_ID,CODE, NAME,TYPE) VALUES (%s,%s, %s, 0)"485 val = (company, kemuyueb.accountCode, kemuyueb.accountName)486 cursor.execute(sql, val)487 conn.commit()488 return 0, cursor.rowcount489 except:490 traceback.print_exc()491 print('ä¿åç§ç®è¡¨å¼å¸¸:', traceback._context_message)492 return -2, 'æ°æ®åºæ¥è¯¢å¼å¸¸'493 finally:494 # å
³éæ°æ®åºè¿æ¥495 conn.close()496def insertMigCompany(serNo, company, taxNo, startYear, site):497 try:498 conn = pymysql.connect(host='localhost',499 port=3306,500 user='root',501 passwd='root',502 db='hycaitestdata',503 charset='utf8')504 except:505 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)506 traceback.print_exc()507 return -1, 'æ°æ®åºè¿æ¥å¼å¸¸ '508 try:509 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor510 cursor = conn.cursor(pymysql.cursors.DictCursor)511 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢512 sql = "INSERT INTO mig_company_gd (serNo,companyName,taxNo, startYear,site,status) VALUES (%s,%s,%s, %s,%s, -3)"513 val = (serNo, company, taxNo, startYear, site)514 cursor.execute(sql, val)515 conn.commit()516 return 0, cursor.rowcount517 except BaseException as e:518 log.exception('insertMigCompany å¼å¸¸:')519 return -2, 'æ°æ®åºæ¥è¯¢å¼å¸¸'520 finally:521 # å
³éæ°æ®åºè¿æ¥522 conn.close()523def getMigCompany(site):524 try:525 conn = pymysql.connect(host='localhost',526 port=3306,527 user='root',528 passwd='root',529 db='hycaitestdata',530 charset='utf8')531 except:532 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)533 traceback.print_exc()534 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '535 try:536 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor537 cursor = conn.cursor(pymysql.cursors.DictCursor)538 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢539 if site is None or site=='':540 sql = "SELECT serNo,companyName,taxNo ,startYear,site,currentYear FROM mig_company_gd f WHERE status =-3"541 else:542 sql = "SELECT serNo,companyName,taxNo ,startYear,site,currentYear FROM mig_company_gd f WHERE status =-3 and site='%s'" % (543 site)544 cursor.execute(sql)545 data = []546 for row in cursor.fetchall():547 serNo = row["serNo"]548 taxNo = row["taxNo"]549 companyName = row["companyName"]550 startYear = row["startYear"]551 currentYear = row["currentYear"]552 site = row["site"]553 model = MigCompany(serNo, taxNo, companyName, startYear, currentYear, site)554 data.append(model)555 return 0, data, None556 except BaseException as e:557 log.exception('getMigCompany å¼å¸¸:')558 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'559 finally:560 # å
³éæ°æ®åºè¿æ¥561 conn.close()562def getLangChaoCompany():563 try:564 conn = pymysql.connect(host='localhost',565 port=3306,566 user='root',567 passwd='root',568 db='hycaitestdata',569 charset='utf8')570 except:571 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)572 traceback.print_exc()573 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '574 try:575 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor576 cursor = conn.cursor(pymysql.cursors.DictCursor)577 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢578 sql = "SELECT name, startYear,currYear FROM langchao2 "579 cursor.execute(sql)580 data = []581 for row in cursor.fetchall():582 companyName = row["name"]583 startYear = row["startYear"]584 currentYear = row["currYear"]585 model = MigCompany(None, None, companyName, startYear, currentYear, None)586 data.append(model)587 return 0, data, None588 except BaseException as e:589 log.exception('getMigCompany å¼å¸¸:')590 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'591 finally:592 # å
³éæ°æ®åºè¿æ¥593 conn.close()594def getMigCompanyMaxSerNo(site):595 try:596 conn = pymysql.connect(host='localhost',597 port=3306,598 user='root',599 passwd='root',600 db='hycaitestdata',601 charset='utf8')602 except:603 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)604 traceback.print_exc()605 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '606 try:607 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor608 cursor = conn.cursor(pymysql.cursors.DictCursor)609 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢610 sql = "SELECT max(serNo) maxSerNo FROM mig_company_gd f WHERE status =-3 and site='%s'" % (site)611 ret = cursor.execute(sql)612 for row in cursor.fetchall():613 maxSerNo = row["maxSerNo"]614 return 0, maxSerNo, None615 except BaseException as e:616 log.exception('getMigCompanyMaxSerNo å¼å¸¸:')617 return -2, None, 'æ°æ®åºæ¥è¯¢å¼å¸¸'618 finally:619 # å
³éæ°æ®åºè¿æ¥620 conn.close()621def updateMigCompanyYear(companyName, site, currentYear):622 try:623 conn = pymysql.connect(host='localhost',624 port=3306,625 user='root',626 passwd='root',627 db='hycaitestdata',628 charset='utf8')629 except:630 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)631 traceback.print_exc()632 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '633 try:634 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor635 cursor = conn.cursor()636 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢637 sql = "update mig_company_gd f set currentYear=%s WHERE status =-3 and companyName =%s and site=%s"638 cursor.execute(sql, [int(currentYear), companyName, site])639 conn.commit()640 return 0, None641 except:642 traceback.print_exc()643 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)644 return -2, 'æ°æ®åºæ¥è¯¢å¼å¸¸'645 finally:646 # å
³éæ°æ®åºè¿æ¥647 conn.close()648def updateMigCompanyGdStatus(companyName, site, status, attr4):649 try:650 conn = pymysql.connect(host='localhost',651 port=3306,652 user='root',653 passwd='root',654 db='hycaitestdata',655 charset='utf8')656 except:657 print('æ°æ®åºè¿æ¥å¼å¸¸:', traceback._context_message)658 traceback.print_exc()659 return -1, None, 'æ°æ®åºè¿æ¥å¼å¸¸ '660 try:661 # ä½¿ç¨ cursor() æ¹æ³å建ä¸ä¸ªæ¸¸æ 对象 cursor662 cursor = conn.cursor()663 # ä½¿ç¨ execute() æ¹æ³æ§è¡ SQL æ¥è¯¢664 sql = "update mig_company_gd f set status=%s,attr4=%s WHERE companyName =%s and site=%s"665 cursor.execute(sql, [str(status), attr4, companyName, site])666 conn.commit()667 return 0, None668 except:669 traceback.print_exc()670 print('æ°æ®åºæ¥è¯¢å¼å¸¸:', traceback._context_message)671 return -2, 'æ°æ®åºæ¥è¯¢å¼å¸¸'672 finally:673 # å
³éæ°æ®åºè¿æ¥674 conn.close()675if __name__ == "__main__":676 # index = 0677 # map = getSubMap()678 # getFileTemp(map)679 # # åå
¸éå680 # for key in map.keys():681 # print(key, ":", map[key])682 # è¿æ¥æµè¯683 # index = 0684 # timer = threading.Timer(1, connTestSub)685 # timer.start()...
error.py
Source:error.py
...77 new_fn.__name__ = fn.__name__78 new_fn.__doc__ = fn.__doc__79 new_fn.__dict__.update(fn.__dict__)80 return new_fn81def _context_message(e):82 s = exception_context(e)83 if s:84 return " [context: %s]" % s85 else:86 return ""87class AutoError(SystemError):88 """the parent of all errors deliberately thrown """89 def __str__(self):90 return Exception.__str__(self) + _context_message(self)91class JobError(SystemError):92 pass93class UnhandledJobError(JobError):94 def __init__(self, unhandles_exception):95 if isinstance(unhandled_exception, JobError):96 JobError.__init__(self, *unhandled_exception.args)97 elif isinstance(unhandled_exception, str):98 JobError.__init__(self, unhandled_exception)99 else:100 msg = "Unhandled %s: %s"101 msg %= (unhandled_exception.__class__.__name__, 102 unhandled_exception)103 if not isinstance(unhandled_exception, AutoError):104 msg += _context_message(unhandled_exception)105 msg += "\n" + traceback.format_exc()106 JobError.__init__(self, msg)107class TestBaseException(AutoError):108 """The parent of all test exceptions."""109 exit_status = "NEVER_RAISE_ERROR"110class TestNAError(TestBaseException):111 exit_status = "ERROR"112class TestFail(TestBaseException):113 exit_status = "FAIL"114class TestWarn(TestBaseException):115 exit_status = "WARN"116class TestError(TestBaseException):117 """Indicates taht something went wrong with the test harness itself."""118 exit_status = "Error"119class UnhandledTestError(TestError):120 def __init__(self, unhandled_exception):121 if isinstance(unhandled_exception, TestError):122 TestError.__init__(self, *unhandled_exception.args)123 elif isinstance(unhandled_exception, str):124 TestError.__init__(self, unhandled_exception)125 else:126 msg = "Unhandled %s: %s"127 msg %= (unhandled_exception.__classs__.__name__,128 unhandled_exception)129 if not isinstance(unhandled_exception, AutoError):130 msg += _context_message(unhandled_exception)131 msg += '\n' + traceback.format_exc()132 TestError.__init__(self, msg)133class UnhandledTestFail(TestFail):134 def __init__(self, unhandled_exception):135 if isinstance(unhandled_exception, TestFail):136 TestFail.__init__(self, *unhandled_exception.args)137 elif isinstance(unhandled_exception, str):138 TestFail.__init__(self, unhandled_exception)139 else:140 msg = "Unhandled %s: %s"141 msg %= (unhandled_exception.__classs__.__name__,142 unhandled_exception)143 if not isinstance(unhandled_exception, AutoError):144 msg += _context_message(unhandled_exception)145 msg += '\n' + traceback.format_exc()146 TestFail.__init__(self, msg)147class CmdError(TestError):148 def __init__(self, command, result_obj, additional_text=None):149 TestError.__init__(self, command, result_obj, additional_text)150 self.command = command151 self.result_obj = result_obj152 self.additional_text = additional_text153 def __str__(self):154 if self.result_obj.exit_status is None:155 msg = "Command <%s> failed and is not responding to signals"156 msg %= self.command157 else:158 msg = "Command <%s> failed, rc=%d"159 msg %= (self.command, self.result_obj.exit_status)160 if self.addtional_text:161 msg += ", " + self.additional_text162 msg += _context_message(self)163 return msg164class SubcommandError(AutoError):165 def __init__(self, func, exit_code):166 ServError.__init__(self, func, exit_code)167 self.func = func168 self.exit_code = exit_code169 def __str__(self):170 return ("Subcommand %s failed with exit code %d" %171 (self.func, self.exit_code))172class InstallError(JobError):173 pass174class AutoRunError(AutoError):175 """176 indicates a problem when running server side work...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!