当前位置：首页 > article >正文

pyqt SQL Server 数据库查询-优化

article 2025/4/1 20:18:03

在这里插入图片描述

一、概述

“SQL Server数据库查询工具” 是一款基于Python和PyQt6开发的应用程序，旨在为用户提供便捷的SQL Server数据库管理与查询功能。该工具允许用户连接至SQL Server数据库，浏览数据库中的表结构，执行自定义查询，并以直观的表格形式展示查询结果。同时，工具具备安全防护机制，防止用户执行危险的数据库操作，如删除表或数据库。

二、功能模块

数据库连接：应用程序通过特定连接字符串DRIVER={SQL Server};SERVER=LEGENDLI;DATABASE=testbase;UID=sa;PWD=1连接至SQL Server数据库。连接成功后，自动获取数据库中所有基表的名称，并展示在表列表区域。
表与字段展示：用户在表列表中点击某个表名时，应用程序查询该表的所有字段名，并在字段列表区域展示。表列表和字段列表均设置了选中项整行变蓝的样式，提升用户交互体验。
基本查询：用户在查询值输入框中输入值，并结合所选表名和字段名，点击“查询”按钮，应用程序生成并执行查询语句。若未输入查询值或未选择字段，将执行SELECT *语句查询所选表的全部数据。查询结果以表格形式呈现在结果展示区域。
自定义查询：用户可在自定义查询输入框中输入SQL查询指令，点击“执行”按钮来执行查询。系统会对输入指令进行检查，若包含“drop table”或“drop database”等危险指令，将弹出错误提示，阻止执行。
错误处理：在数据库连接、获取表字段信息、执行查询等操作过程中，若发生异常，应用程序会弹出错误提示弹窗，展示详细的错误信息，帮助用户快速定位和解决问题。

三、界面设计

布局：整体采用垂直布局（QVBoxLayout），从上至下依次为表列表、字段列表与查询值输入框区域，自定义查询输入框区域，以及查询结果展示区域。表列表、字段列表与查询值输入框位于同一水平布局（QHBoxLayout）内，方便用户操作。
样式：窗口背景色设置为#f0f0f0，营造简洁舒适的视觉效果。标签（QLabel）字体加粗，增强辨识度。按钮（QPushButton）背景色为#4CAF50，鼠标悬停时变为#45a049，按钮样式为圆角矩形，内边距设置使按钮美观且易于点击。输入框（QLineEdit）设置了1px的边框和圆角，提升界面整体质感。

四、技术实现

编程语言：Python作为主要开发语言，利用其丰富的库和简洁的语法，高效实现数据库连接、界面交互和业务逻辑处理。
数据库连接：借助pyodbc库建立与SQL Server数据库的连接，通过执行SQL语句完成数据查询和元数据获取操作。
图形界面：基于PyQt6库构建图形用户界面，使用各种Qt组件（如QWidget、QVBoxLayout、QHBoxLayout、QListWidget、QLineEdit、QPushButton、QTableWidget等）实现界面布局和用户交互功能。通过设置样式表（setStyleSheet）对界面元素进行美化，利用信号与槽机制（如itemClicked.connect、clicked.connect）实现用户操作的响应。
异常处理：在关键操作代码块中使用try - except语句捕获异常，并通过自定义的show_error_message方法弹出错误提示弹窗，确保应用程序的稳定性和可靠性。

五、使用说明

环境要求：确保系统已安装Python环境以及所需的pyodbc、PyQt6库。
启动应用：运行Python脚本，启动数据库查询工具。
连接数据库：应用程序自动尝试连接指定的SQL Server数据库。若连接失败，根据错误提示检查数据库连接配置。
查询操作：在表列表中选择表，字段列表将自动更新显示该表的字段。在查询值输入框输入值，结合所选表和字段，点击“查询”按钮进行基本查询；或在自定义查询输入框输入SQL指令，点击“执行”按钮进行自定义查询。查询结果将显示在下方表格中。
错误处理：若操作过程中出现错误，仔细阅读错误提示弹窗中的信息，根据提示解决问题，如检查语法错误、确认数据库对象是否存在等。

代码

import sys
import pyodbc
from PyQt6.QtWidgets import QApplication, QWidget, QVBoxLayout, QHBoxLayout, QListWidget, QLineEdit, QPushButton, \
    QTableWidget, QTableWidgetItem, QLabel, QMessageBox
from PyQt6.QtGui import QFont
from PyQt6.QtCore import Qt


class DatabaseQueryApp(QWidget):
    def __init__(self):
        super().__init__()
        self.initUI()
        self.connect_to_database()

    def initUI(self):
        # 设置窗口字体
        font = QFont()
        font.setPointSize(12)
        self.setFont(font)

        # 创建布局
        main_layout = QVBoxLayout()

        # 表列表、字段列表和输入框布局
        list_input_layout = QHBoxLayout()

        # 表列表部分
        table_label = QLabel("表列表:")
        list_input_layout.addWidget(table_label)
        self.table_list = QListWidget()
        # 调整显示表的控件大小
        self.table_list.setFixedHeight(100)
        # 设置 QListWidget 选中项整行变蓝色
        self.table_list.setStyleSheet("QListWidget::item:selected { background-color: blue; color: white; }")
        self.table_list.itemClicked.connect(self.show_table_columns)
        list_input_layout.addWidget(self.table_list)

        # 字段列表部分
        column_label = QLabel("字段列表:")
        list_input_layout.addWidget(column_label)
        self.column_list = QListWidget()
        self.column_list.setFixedHeight(100)
        self.column_list.setStyleSheet("QListWidget::item:selected { background-color: blue; color: white; }")
        list_input_layout.addWidget(self.column_list)

        # 查询输入部分
        value_label = QLabel("查询值:")
        list_input_layout.addWidget(value_label)
        self.value_input = QLineEdit()
        self.query_button = QPushButton('查询')
        # 获取当前按钮的大小
        current_width = self.query_button.sizeHint().width()
        current_height = self.query_button.sizeHint().height()
        # 设置按钮大小为原来的两倍
        button_size = (current_width * 2, current_height * 2)
        self.query_button.setFixedSize(*button_size)
        self.query_button.clicked.connect(self.execute_query)
        list_input_layout.addWidget(self.value_input)
        list_input_layout.addWidget(self.query_button)

        main_layout.addLayout(list_input_layout)

        # 自定义查询输入部分
        custom_query_layout = QHBoxLayout()
        custom_query_label = QLabel("自定义查询:")
        custom_query_layout.addWidget(custom_query_label)
        self.custom_query_input = QLineEdit()
        self.custom_execute_button = QPushButton('执行')
        self.custom_execute_button.setFixedSize(*button_size)
        self.custom_execute_button.clicked.connect(self.execute_custom_query)
        custom_query_layout.addWidget(self.custom_query_input)
        custom_query_layout.addWidget(self.custom_execute_button)

        main_layout.addLayout(custom_query_layout)

        # 查询结果表格部分
        result_label = QLabel("查询结果:")
        main_layout.addWidget(result_label)
        self.result_table = QTableWidget()
        self.result_table.setColumnCount(0)
        self.result_table.setRowCount(0)
        # 设置 QTableWidget 选择行为为整行选择
        self.result_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
        # 设置 QTableWidget 选中行整行变蓝色
        self.result_table.setStyleSheet("QTableWidget::item:selected { background-color: blue; color: white; }")

        main_layout.addWidget(self.result_table)

        self.setLayout(main_layout)
        self.setWindowTitle('数据库查询工具')
        self.setGeometry(300, 300, 1000, 800)
        self.setStyleSheet("""
            QWidget {
                background-color: #f0f0f0;
            }
            QLabel {
                font-weight: bold;
            }
            QPushButton {
                background-color: #4CAF50;
                color: white;
                padding: 10px 20px;
                border: none;
                border-radius: 5px;
            }
            QPushButton:hover {
                background-color: #45a049;
            }
            QLineEdit {
                padding: 8px;
                border: 1px solid #ccc;
                border-radius: 5px;
            }
        """)
        self.show()

    def connect_to_database(self):
        try:
            # 使用指定的连接字符串
            connection_string = 'DRIVER={SQL Server};SERVER=LEGENDLI;DATABASE=testbase;UID=sa;PWD=1'
            self.conn = pyodbc.connect(connection_string)
            cursor = self.conn.cursor()
            cursor.execute("SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE = 'BASE TABLE'")
            tables = cursor.fetchall()
            for table in tables:
                self.table_list.addItem(table[0])
        except Exception as e:
            self.show_error_message(f"数据库连接错误: {e}")

    def show_table_columns(self, item):
        table_name = item.text()
        try:
            cursor = self.conn.cursor()
            cursor.execute(f"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '{table_name}'")
            columns = cursor.fetchall()
            self.column_list.clear()
            for column in columns:
                self.column_list.addItem(column[0])
        except Exception as e:
            self.show_error_message(f"获取字段信息错误: {e}")

    def execute_query(self):
        selected_table_items = self.table_list.selectedItems()
        selected_column_items = self.column_list.selectedItems()
        if not selected_table_items:
            self.show_error_message("请选择一个表。")
            return

        table_name = selected_table_items[0].text()
        value = self.value_input.text()

        if not selected_column_items or not value:
            query = f"SELECT * FROM {table_name}"
        else:
            column_name = selected_column_items[0].text()
            query = f"SELECT * FROM {table_name} WHERE {column_name} = '{value}'"

        try:
            cursor = self.conn.cursor()
            cursor.execute(query)
            results = cursor.fetchall()
            headers = [description[0] for description in cursor.description]

            self.result_table.setColumnCount(len(headers))
            self.result_table.setRowCount(len(results))
            self.result_table.setHorizontalHeaderLabels(headers)

            for row_index, row_data in enumerate(results):
                for col_index, col_data in enumerate(row_data):
                    item = QTableWidgetItem(str(col_data))
                    self.result_table.setItem(row_index, col_index, item)

        except Exception as e:
            self.show_error_message(f"查询错误: {e}")

    def execute_custom_query(self):
        query = self.custom_query_input.text().strip().lower()
        # 检查是否包含危险指令
        if "drop table" in query or "drop database" in query:
            self.show_error_message("不允许执行删除表或删除数据库的指令。")
            return

        try:
            cursor = self.conn.cursor()
            cursor.execute(query)
            results = cursor.fetchall()
            headers = [description[0] for description in cursor.description]

            self.result_table.setColumnCount(len(headers))
            self.result_table.setRowCount(len(results))
            self.result_table.setHorizontalHeaderLabels(headers)

            for row_index, row_data in enumerate(results):
                for col_index, col_data in enumerate(row_data):
                    item = QTableWidgetItem(str(col_data))
                    self.result_table.setItem(row_index, col_index, item)

        except Exception as e:
            self.show_error_message(f"查询错误: {e}")

    def show_error_message(self, message):
        msg_box = QMessageBox()
        msg_box.setIcon(QMessageBox.Icon.Critical)
        msg_box.setText(message)
        msg_box.setWindowTitle("错误提示")
        msg_box.exec()


if __name__ == '__main__':
    app = QApplication(sys.argv)
    ex = DatabaseQueryApp()
    sys.exit(app.exec())