我编写了一个 PyQt5 应用程序，用于在 SQL 数据库中搜索用户输入。一切正常，但我没有运气尝试添加一个工作线程来处理搜索过程并防止应用程序变得无响应。我正在使用 QT Designer 为 UI 生成代码，该代码由主窗口、用于输入的 QLineEdit 框和用于开始搜索的 QPushButton 组成。至于进度条（和用于停止搜索的取消按钮），我正在考虑两个选项：将其“锚定”在主窗口中，或者将其弹出在对话框中。在 QMainWindow 类中，我定义了一个获取用户输入的函数（set_validators）和一个处理搜索并显示结果的函数（save_pushed）。我了解信号/插槽的基础，并且查看了几个示例，但我不确定如何将它们组合在一起。部分问题是我认为的可变范围。任何人都可以提供一般指导吗？
#!/usr/bin/env python
import pyodbc
import sys
from PyQt5.QtWidgets import QMainWindow, QApplication, QTableWidgetItem, QMessageBox
from PyQt5.QtGui import QRegExpValidator
from PyQt5.QtCore import QRegExp
from PyQt5 import uic
from ui_pysearch import Ui_pySearch
class pySearchMain(QMainWindow):
def __init__(self):
QMainWindow.__init__(self)
self.ui = Ui_pySearch()
self.ui.setupUi(self)
self.ui.tableWidget_ResultsIncident.setHorizontalHeaderLabels(["A", "B", "C"])
self.ui.tableWidget_ResultsCompany.setHorizontalHeaderLabels(["A", "B", "C"])
self._set_validators()
self.ui.Button_Search.clicked.connect(self._save_pushed)
def _set_validators(self):
reg_ex = QRegExp(r"[A-Za-z0-9]{0,6}")
val_company_custid = QRegExpValidator(reg_ex, self.ui.Input_lineEdit_CompanyPIN)
self.ui.Input_lineEdit_CompanyPIN.setValidator(val_company_custid)
reg_ex = QRegExp(r"[A-Za-z ]{0,25}")
val_company_name = QRegExpValidator(reg_ex, self.ui.Input_lineEdit_CompanyName)
self.ui.Input_lineEdit_CompanyName.setValidator(val_company_name)
reg_ex = QRegExp(r"[A-Za-z ]{0,25}")
val_incident_service = QRegExpValidator(reg_ex, self.ui.Input_lineEdit_IncidentService)
self.ui.Input_lineEdit_IncidentService.setValidator(val_incident_service)
def _save_pushed(self):
self.ui.tableWidget_ResultsIncident.clearContents()
self.ui.tableWidget_ResultsCompany.clearContents()
Company_CustID = self.ui.Input_lineEdit_CompanyPIN.text()
Company_Name = self.ui.Input_lineEdit_CompanyName.text()
Incident_Service = self.ui.Input_lineEdit_IncidentService.text()
Set_Company_RecID = set()
Set_Company_CustID = set()
Set_Company_CustID_Substring = set()
Set_Company_Name = set()
Set_Company_Name_Substring = set()
Set_IncidentNumber_CompanyName = set()
Set_IncidentNumber_Service = set()
Match_Input_lineEdit = 1
Match_RecID_CompanyCustID = -1
Match_RecID_CompanyName = -1
if len(str(Company_CustID)) == 6:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select RecId, CustID, Name from Company")
for row in SearchCursor.fetchall():
if str(row[1]).lower() == str(Company_CustID).lower():
Match_RecID_CompanyCustID = row[0]
Set_Company_RecID.add(str(row[1]))
if len(str(Company_CustID)) > 1:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select RecId, CustID, Name from Company")
for row in SearchCursor.fetchall():
if str(row[1]).lower().rfind(str(Company_CustID).lower()) != -1:
Set_Company_CustID_Substring.add(row[0])
Set_Company_RecID.add(str(row[1]))
if len(str(Company_Name)) > 2:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select RecId, CustID, Name from Company")
for row in SearchCursor.fetchall():
if str(row[2]).lower() == str(Company_Name).lower():
Match_RecID_CompanyName = row[0]
Set_Company_RecID.add(str(row[1]))
if str(row[2]).lower().rfind(str(Company_Name).lower()) != -1:
Set_Company_Name_Substring.add(row[0])
Set_Company_RecID.add(str(row[1]))
if Match_RecID_CompanyCustID != -1:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select IncidentNumber from Incident")
for row in SearchCursor.fetchall():
if row[0] == Match_RecID_CompanyCustID:
Set_Company_CustID.add(int(row[0]))
if Match_RecID_CompanyName != -1:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select IncidentNumber, CompanyLink_RecID from Incident")
for row in SearchCursor.fetchall():
if row[1] == Match_RecID_CompanyName:
Set_Company_Name.add(int(row[0]))
if len(str(Company_Name)) > 2:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select IncidentNumber, CompanyName from Incident")
for row in SearchCursor.fetchall():
if str(row[1]).lower() == str(Company_Name).lower():
Set_IncidentNumber_CompanyName.add(int(row[0]))
if str(row[1]).lower().rfind(str(Company_Name).lower()) != -1:
Set_IncidentNumber_CompanyName.add(int(row[0]))
if len(Incident_Service) > 2:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select IncidentNumber, Service from Incident")
for row in SearchCursor.fetchall():
if (row[1].lower()).rfind(Incident_Service.lower()) != -1:
Set_IncidentNumber_Service.add(int(row[0]))
if len(Set_Company_CustID_Substring) != 0:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select IncidentNumber, CompanyLink_RecID from Incident")
for row in SearchCursor.fetchall():
for Set_Company_CustID_Substring_Item in Set_Company_CustID_Substring:
if row[1] == Set_Company_CustID_Substring_Item:
Set_Company_CustID.add(int(row[0]))
if len(Set_Company_Name_Substring) != 0:
with pyodbc.connect('DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
SearchCursor = DB_Connect.cursor()
SearchCursor.execute("select IncidentNumber, CompanyLink_RecID from Incident")
for row in SearchCursor.fetchall():
for Set_Company_Name_Substring_Item in Set_Company_Name_Substring:
if row[1] == Set_Company_Name_Substring_Item:
Set_Company_Name.add(int(row[0]))
if len(Set_Company_CustID) == 0 and len(Company_CustID) != 0:
Match_Input_lineEdit = 0
if len(Set_Company_Name) == 0 and len(Company_Name) != 0:
Match_Input_lineEdit = 0
if len(Set_IncidentNumber_Service) == 0 and len(Incident_Service) != 0:
Match_Input_lineEdit = 0
if len(Set_IncidentNumber_CompanyName) == 0 and len(Company_Name) != 0:
Match_Input_lineEdit = 0
Match_Input_lineEdit = 0
if Match_Input_lineEdit == 0:
alert = QMessageBox()
alert.setText('No results')
self.ui.tableWidget_ResultsIncident.clearContents()
self.ui.tableWidget_ResultsIncident.setRowCount(0)
alert.exec_()
return
List_Of_All_Sets = [Set_Company_CustID, Set_Company_Name, Set_IncidentNumber_Service, Set_IncidentNumber_CompanyName]
try:
Set_Intersection = set.intersection(*(s for s in List_Of_All_Sets if s))
if len(Set_Intersection) == 0:
alert = QMessageBox()
alert.setText('No results')
self.ui.tableWidget_ResultsIncident.clearContents()
self.ui.tableWidget_ResultsIncident.setRowCount(0)
alert.exec_()
except TypeError:
alert = QMessageBox()
alert.setText('No results')
self.ui.tableWidget_ResultsIncident.clearContents()
self.ui.tableWidget_ResultsIncident.setRowCount(0)
alert.exec_()
else:
if self.ui.radioButton_Search_Incident.isChecked():
self.ui.tableWidget_ResultsIncident.setRowCount(0)
tableWidget_ResultsIncident_row_num = self.ui.tableWidget_ResultsIncident.rowCount()
with pyodbc.connect(
'DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
Set_Intersection_Cursor = DB_Connect.cursor()
Set_Intersection_Cursor.execute("select CompanyLink_RecID, IncidentNumber from Incident")
for Set_Intersection_Cursor_Incident_Item in Set_Intersection_Cursor.fetchall():
for Set_Intersection_Item in Set_Intersection:
if Set_Intersection_Cursor_Incident_Item[1] == Set_Intersection_Item:
Match_Intersection_RecID = Set_Intersection_Cursor_Incident_Item[0]
Set_Intersection_Cursor.execute("select RecId, CustID, Name from Company")
for Set_Intersection_Cursor_Company_Item in Set_Intersection_Cursor.fetchall():
if Set_Intersection_Cursor_Company_Item[0] == Match_Intersection_RecID:
self.ui.tableWidget_ResultsIncident.insertRow(tableWidget_ResultsIncident_row_num)
self.ui.tableWidget_ResultsIncident.setItem(tableWidget_ResultsIncident_row_num, 0, QTableWidgetItem(str(Set_Intersection_Item)))
self.ui.tableWidget_ResultsIncident.setItem(tableWidget_ResultsIncident_row_num, 1, QTableWidgetItem(str(Set_Intersection_Cursor_Company_Item[1])))
Set_Intersection_Cursor.execute("select IncidentNumber, Priority, Service, TypeOfIncident, CellNum, CreatedDateTime, Status, TACPrime, Owner from Incident")
for row in Set_Intersection_Cursor.fetchall():
if row[0] == Set_Intersection_Item:
self.ui.tableWidget_ResultsIncident.setItem(tableWidget_ResultsIncident_row_num, 2, QTableWidgetItem(str(row[1]))) #Priority
if row[0] == Set_Intersection_Item:
self.ui.tableWidget_ResultsIncident.setItem(tableWidget_ResultsIncident_row_num, 3, QTableWidgetItem(str(row[2]))) #Service
if len(Set_Company_RecID) != 0 and self.ui.radioButton_Search_Company.isChecked():
self.ui.tableWidget_ResultsCompany.setColumnWidth(0, 215)
self.ui.tableWidget_ResultsCompany.setColumnWidth(1, 215)
self.ui.tableWidget_ResultsCompany.setColumnWidth(2, 215)
self.ui.tableWidget_ResultsCompany.setRowCount(0)
tableWidget_ResultsCompany_row_num = self.ui.tableWidget_ResultsCompany.rowCount()
with pyodbc.connect(
'DRIVER={ODBC Driver 17 for SQL Server};SERVER=ServerIP;DATABASE=Sqldbname;UID=db_username;PWD=db_password') as DB_Connect:
Company_RecID_Cursor = DB_Connect.cursor()
Company_RecID_Cursor.execute("select RecID, CustID, Name from Company")
for Company_RecID_Cursor_Item in Company_RecID_Cursor.fetchall():
for Set_Company_RecID_Item in Set_Company_RecID:
if Company_RecID_Cursor_Item[1] == Set_Company_RecID_Item:
self.ui.tableWidget_ResultsCompany.insertRow(tableWidget_ResultsCompany_row_num)
self.ui.tableWidget_ResultsCompany.setItem(tableWidget_ResultsCompany_row_num, 0, QTableWidgetItem(str(Set_Company_RecID_Item)))
self.ui.tableWidget_ResultsCompany.setItem(tableWidget_ResultsCompany_row_num, 1, QTableWidgetItem(str(Company_RecID_Cursor_Item[2])))
self.ui.tableWidget_ResultsCompany.setItem(tableWidget_ResultsCompany_row_num, 2, QTableWidgetItem(str(Company_RecID_Cursor_Item[0])))
if __name__ == '__main__':
app = QApplication(sys.argv)
main = pySearchMain()
main.show()
sys.exit(app.exec_())