PyQt5结合QAbstractTableModel从数据库加载数据时显示加载GIF的实现方案

Hey there! I totally get where you're coming from—QThread can feel super confusing when you're just starting out with PyQt5, especially when you're trying to keep your UI responsive while loading big datasets from a database. Let's fix this together with a proper thread-based solution, no QTimer required.

First, let's go over the core ideas we'll use:

• Offload heavy work to a separate thread: Database queries are slow, so we'll run them in a background thread to keep your UI from freezing.
• Use signal-slot communication: Threads can't directly update the UI, so we'll use signals to send data and status updates from the background thread to the main UI thread.
• Manage the loading state: We'll show a GIF spinner and status text while loading, then automatically hide them once the data is ready.

Complete Working Code #

Here's the full code that implements everything you need:

from PyQt5 import QtCore, QtWidgets, QtGui
import pandas as pd
import numpy as np
import pyodbc

class NumpyArrayModel(QtCore.QAbstractTableModel):
    def __init__(self, array, headers, parent=None):
        super().__init__(parent)
        self._array = array
        self._headers = headers
        self.r, self.c = np.shape(self._array)

    @property
    def array(self):
        return self._array

    @property
    def headers(self):
        return self._headers

    def rowCount(self, parent=QtCore.QModelIndex()):
        return self.r

    def columnCount(self, parent=QtCore.QModelIndex()):
        return self.c

    def headerData(self, p_int, orientation, role):
        if role == QtCore.Qt.DisplayRole:
            if orientation == QtCore.Qt.Horizontal:
                if p_int < len(self.headers):
                    return self.headers[p_int]
            elif orientation == QtCore.Qt.Vertical:
                return p_int + 1
        return None

    def data(self, index, role=QtCore.Qt.DisplayRole):
        if not index.isValid():
            return None
        row = index.row()
        column = index.column()
        if row < 0 or row >= self.rowCount() or column < 0 or column >= self.columnCount():
            return None
        if role == QtCore.Qt.DisplayRole:
            return str(self.array[row, column])
        return None

    def setData(self, index, value, role):
        if not index.isValid() or role != QtCore.Qt.EditRole:
            return False
        row = index.row()
        column = index.column()
        if row < 0 or row >= self.rowCount() or column < 0 or column >= self.columnCount():
            return False
        self._array[row][column] = value
        self.dataChanged.emit(index, index)
        return True

class DatabaseWorker(QtCore.QObject):
    # Signals: send data to UI thread, notify when done
    send_data = QtCore.pyqtSignal(pd.DataFrame, list)
    finished = QtCore.pyqtSignal()

    def __init__(self, server, database, username, passwd, query):
        super().__init__()
        self.server = server
        self.database = database
        self.username = username
        self.passwd = passwd
        self.query = query

    def run(self):
        # This runs in the background thread
        try:
            # Connect to database and load data
            conn = pyodbc.connect(
                f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={self.server};DATABASE={self.database};UID={self.username};PWD={self.passwd}'
            )
            df = pd.read_sql_query(self.query, conn)
            headers = df.columns.tolist()
            # Send data back to UI thread
            self.send_data.emit(df, headers)
        except Exception as e:
            # You can add error handling here (e.g., emit an error signal)
            print(f"Error loading data: {e}")
        finally:
            # Notify UI thread we're done
            self.finished.emit()

class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)
        self.init_ui()

    def init_ui(self):
        vLayout = QtWidgets.QVBoxLayout(self)
        hLayout = QtWidgets.QHBoxLayout()

        # Status label and load button
        self.pathLE = QtWidgets.QLabel(self)
        hLayout.addWidget(self.pathLE)
        self.loadBtn = QtWidgets.QPushButton("Load data", self)
        hLayout.addWidget(self.loadBtn)
        vLayout.addLayout(hLayout)

        # Table view
        self.pandasTv = QtWidgets.QTableView(self)
        self.pandasTv.setSortingEnabled(True)
        vLayout.addWidget(self.pandasTv)

        # Loading GIF setup
        self.loading_label = QtWidgets.QLabel(self)
        self.loading_label.setAlignment(QtCore.Qt.AlignCenter)
        # Replace with your own loading GIF path
        self.loading_movie = QtGui.QMovie("loading_spinner.gif")
        self.loading_label.setMovie(self.loading_movie)
        self.loading_label.hide()
        vLayout.addWidget(self.loading_label)

        # Connect button click to load function
        self.loadBtn.clicked.connect(self.start_loading_data)

    def start_loading_data(self):
        # Update UI to show loading state
        self.pathLE.setText("Loading data...")
        self.loadBtn.setEnabled(False)
        self.loading_label.show()
        self.loading_movie.start()

        # Database connection details (replace with your own)
        server = '190.11.71.09'
        database = ''
        username = 'Admin'
        passwd = ''
        query = "select * from database.dbo.tableName(nolock)"

        # Create worker and thread
        self.worker = DatabaseWorker(server, database, username, passwd, query)
        self.thread = QtCore.QThread()

        # Move worker to the thread
        self.worker.moveToThread(self.thread)

        # Connect signals and slots
        # Start worker's run method when thread starts
        self.thread.started.connect(self.worker.run)
        # Update table when worker sends data
        self.worker.send_data.connect(self.update_table)
        # Clean up thread when worker finishes
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        # Hide loading state when done
        self.worker.finished.connect(self.hide_loading_state)

        # Start the thread
        self.thread.start()

    def update_table(self, df, headers):
        # This runs in the main UI thread - safe to update UI here
        array = np.array(df.values)
        model = NumpyArrayModel(array, headers)
        self.pandasTv.setModel(model)

    def hide_loading_state(self):
        # Reset UI to normal state
        self.pathLE.setText("Data loaded successfully!")
        self.loadBtn.setEnabled(True)
        self.loading_movie.stop()
        self.loading_label.hide()

if __name__ == "__main__":
    import sys
    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())

Let's Break Down the Code #

1. The DatabaseWorker Class #

This is where our heavy lifting happens. Instead of inheriting from QThread, we create a regular QObject and move it to a QThread—this is the recommended way to use threads in PyQt because it keeps your code flexible.

• Signals:
  • send_data: Sends the loaded DataFrame and column headers back to the UI thread.
  • finished: Tells the UI thread we're done loading data.
• run() Method: This is executed in the background thread. It connects to the database, runs the query, and emits the appropriate signals. If there's an error, you can add error handling here (like emitting an error signal to show a message box).

2. UI Setup in Widget #

• Loading GIF: We add a QLabel to display the spinner, using QMovie to animate the GIF. It's hidden by default.
• start_loading_data() Method:
  • First, we update the UI to show the loading state: set the status text, disable the load button, show the GIF, and start the animation.
  • We create the worker and thread, then move the worker to the thread (this is crucial—all the worker's methods will now run in the thread).
  • We connect all the signals: when the thread starts, it triggers the worker's run() method; when the worker sends data, we update the table; when the worker finishes, we clean up the thread and reset the UI.

3. Updating the Table Safely #

The update_table() method is called via the send_data signal, which means it runs in the main UI thread. This is important because PyQt doesn't allow UI operations (like setting a table model) from background threads. We convert the DataFrame to a numpy array and create our NumpyArrayModel here.

4. Thread Cleanup #

We connect the worker's finished signal to the thread's quit() method, and both the worker and thread to deleteLater()—this ensures we don't leave orphaned threads or objects hanging around, which can cause memory leaks.

Important Notes #

• Replace the GIF Path: Make sure to replace "loading_spinner.gif" with the actual path to your loading animation file. You can find free loading GIFs online if you don't have one.
• Database Details: Update the server, database, username, password, and query to match your own setup.
• Error Handling: I added a basic try-except block in the worker—you can expand this to emit an error signal and show a message box in the UI if something goes wrong.

内容的提问来源于stack exchange，提问作者Rimona