dadidanlab.com

Accelerating PyQt6 Kanban Applications: Optimizing Database-Driven UI Loading

Developing a PyQt6 application that loads extensive data from a database, splits it into categories, sorts it, and then constructs complex UI elements like job tiles, often encounters performance bottlenecks, particularly slow loading times for work centers. The primary solution involves offloading data-intensive operations to a separate thread and optimizing UI rendering.

The Problem

A common challenge in PyQt6 applications interfacing with databases is that long-running database queries, data processing (splitting, sorting), and subsequent UI widget construction (e.g., numerous QMenu, QObject-derived custom job tiles) occur on the main thread. This leads to an unresponsive application UI, frozen windows, and a poor user experience during startup or data refresh operations, especially when dealing with a large volume of job orders across multiple work centers.

The Solution

To significantly improve application responsiveness and loading speed, data retrieval and heavy processing must be performed asynchronously on a dedicated worker thread. Once the data is ready, it is then safely passed back to the main UI thread for efficient rendering.

import sys
import time
import random
from datetime import datetime, timedelta
from PyQt6.QtWidgets import (QApplication, QWidget, QVBoxLayout, QPushButton,
                             QLabel, QListWidget, QListWidgetItem, QProgressBar)
from PyQt6.QtCore import (QObject, QThread, pyqtSignal, Qt)

# Simulate a Job Tile structure
class JobTileData:
    def __init__(self, job_id, title, work_center, due_date):
        self.job_id = job_id
        self.title = title
        self.work_center = work_center
        self.due_date = due_date

    def __repr__(self):
        return f"Job {self.job_id}: {self.title} (WC: {self.work_center}, Due: {self.due_date.strftime('%Y-%m-%d')})"

# Worker class to perform database operations in a separate thread
class DatabaseWorker(QObject):
    data_ready = pyqtSignal(list)
    progress_update = pyqtSignal(int)
    error_occurred = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)
        self._is_running = True

    def run(self):
        try:
            # Simulate database connection and data fetching
            # Replace with actual DB calls (e.g., SQLAlchemy, psycopg2, sqlite3)
            print("Simulating database query...")
            total_jobs = 100
            jobs_data = []

            for i in range(total_jobs):
                if not self._is_running:
                    return # Allow early termination

                # Simulate a delay for each row fetch
                time.sleep(0.01) 

                job_id = f"J{i:04d}"
                title = f"Task {i} for Project X"
                work_center = f"WC-{random.randint(1, 5)}" # Simulate 5 work centers
                due_date = datetime.now() + timedelta(days=random.randint(1, 30))

                jobs_data.append(JobTileData(job_id, title, work_center, due_date))
                self.progress_update.emit(int((i + 1) / total_jobs * 100))

            # Simulate splitting into work centers and sorting
            work_center_jobs = {}
            for job in jobs_data:
                work_center_jobs.setdefault(job.work_center, []).append(job)

            # Sort jobs within each work center by due date
            for wc, jobs in work_center_jobs.items():
                work_center_jobs[wc] = sorted(jobs, key=lambda x: x.due_date)

            # Convert dictionary of lists back to a flat list for simpler signal emission
            # Or emit a more complex structure if needed by the UI.
            processed_data = []
            for wc_key in sorted(work_center_jobs.keys()): # Ensure consistent order
                processed_data.extend(work_center_jobs[wc_key])

            print(f"Database query complete. Found {len(processed_data)} jobs.")
            self.data_ready.emit(processed_data)

        except Exception as e:
            self.error_occurred.emit(str(e))
        finally:
            print("Worker finished.")

    def stop(self):
        self._is_running = False

# Main Application Window
class KanbanApp(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("PyQt6 Kanban Loader (Optimized)")
        self.setGeometry(100, 100, 800, 600)

        self.layout = QVBoxLayout()
        self.setLayout(self.layout)

        self.load_button = QPushButton("Load Job Orders")
        self.load_button.clicked.connect(self.start_loading)
        self.layout.addWidget(self.load_button)

        self.progress_bar = QProgressBar()
        self.progress_bar.setAlignment(Qt.AlignmentFlag.AlignCenter)
        self.progress_bar.setVisible(False)
        self.layout.addWidget(self.progress_bar)

        self.status_label = QLabel("Click 'Load Job Orders' to fetch data.")
        self.layout.addWidget(self.status_label)

        self.work_center_lists = {} # Dictionary to hold QListWidgets for each work center
        self.current_worker_thread = None
        self.current_database_worker = None

    def start_loading(self):
        self.load_button.setEnabled(False)
        self.status_label.setText("Loading data from database...")
        self.progress_bar.setValue(0)
        self.progress_bar.setVisible(True)

        # Clear existing work center lists if any
        for wc_list in self.work_center_lists.values():
            wc_list.deleteLater()
        self.work_center_lists.clear()

        # Create a QThread and a Worker instance
        self.current_worker_thread = QThread()
        self.current_database_worker = DatabaseWorker()

        # Move the worker to the new thread
        self.current_database_worker.moveToThread(self.current_worker_thread)

        # Connect signals and slots
        self.current_worker_thread.started.connect(self.current_database_worker.run)
        self.current_database_worker.data_ready.connect(self.display_jobs)
        self.current_database_worker.progress_update.connect(self.progress_bar.setValue)
        self.current_database_worker.error_occurred.connect(self.handle_error)

        # Ensure cleanup when thread finishes
        self.current_database_worker.data_ready.connect(self.current_worker_thread.quit)
        self.current_database_worker.data_ready.connect(self.current_worker_thread.deleteLater)
        self.current_database_thread_finished = lambda: self.cleanup_thread(self.current_worker_thread, self.current_database_worker)
        self.current_worker_thread.finished.connect(self.current_database_thread_finished)

        # Start the thread
        self.current_worker_thread.start()

    def display_jobs(self, job_data_list):
        self.status_label.setText("Data loaded. Constructing UI...")
        self.progress_bar.setVisible(False)
        self.load_button.setEnabled(True)

        # Group jobs by work center
        work_center_grouped_jobs = {}
        for job in job_data_list:
            work_center_grouped_jobs.setdefault(job.work_center, []).append(job)

        # Create QListWidgets for each work center and populate them
        for wc_name in sorted(work_center_grouped_jobs.keys()):
            wc_label = QLabel(f"<b>Work Center: {wc_name}</b>")
            wc_list_widget = QListWidget()
            wc_list_widget.setMinimumHeight(150) # Example sizing

            # Add widgets for the new work center
            self.layout.addWidget(wc_label)
            self.layout.addWidget(wc_list_widget)
            self.work_center_lists[wc_name] = wc_list_widget

            for job in work_center_grouped_jobs[wc_name]:
                # For demonstration, a simple QListWidgetItem is used.
                # In a real Kanban, you'd create a custom QWidget (JobTile)
                # and use setItemWidget to embed it.
                item = QListWidgetItem(f"{job.title} (Due: {job.due_date.strftime('%Y-%m-%d')})")
                wc_list_widget.addItem(item)
                # Example for custom widget (uncomment and implement JobTile)
                # custom_tile = JobTile(job) 
                # item.setSizeHint(custom_tile.sizeHint())
                # wc_list_widget.setItemWidget(item, custom_tile)

        self.status_label.setText(f"Loaded {len(job_data_list)} jobs across {len(self.work_center_lists)} work centers.")

    def handle_error(self, message):
        self.status_label.setText(f"Error: {message}")
        self.progress_bar.setVisible(False)
        self.load_button.setEnabled(True)
        self.cleanup_thread(self.current_worker_thread, self.current_database_worker)

    def cleanup_thread(self, thread, worker):
        if thread and thread.isRunning():
            print("Stopping worker and cleaning up thread.")
            worker.stop() # Request worker to stop its loop if applicable
            thread.quit()
            thread.wait(2000) # Wait for thread to finish, with a timeout
            if thread.isRunning():
                thread.terminate() # Force termination if it doesn't stop gracefully
            print("Thread cleanup complete.")

        if worker:
            worker.deleteLater()
        if thread:
            thread.deleteLater()

        self.current_worker_thread = None
        self.current_database_worker = None


if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = KanbanApp()
    window.show()
    sys.exit(app.exec())

Why It Works

  1. Asynchronous Data Loading (QThread, QObject, pyqtSignal):
    • The DatabaseWorker class, inheriting from QObject, encapsulates the database queries and data processing logic.
    • This worker is moved to a separate QThread, preventing these long-running operations from blocking the main GUI thread. The application remains responsive, and users can interact with other parts of the UI or see progress.
    • pyqtSignal is used for safe, thread-agnostic communication. The data_ready, progress_update, and error_occurred signals emitted by the DatabaseWorker are connected to slots in the KanbanApp (main GUI thread). This ensures that UI updates only occur on the main thread, which is a fundamental requirement for thread-safe GUI programming in Qt.
  2. Decoupled UI Construction:
    • Data is fully retrieved and processed in the background before any significant UI construction begins. This ensures that the UI components are built with complete, sorted data, rather than being updated incrementally, which can cause flickering or layout recalculations.
    • For displaying many “job tiles,” consider using QListWidget (as shown, with optional setItemWidget for custom widgets) or QListView with a custom QAbstractListModel and QStyledItemDelegate. This pattern virtualizes rendering, displaying only visible items and drastically reducing the memory and CPU overhead compared to creating individual QFrame or QPushButton instances for every single job tile.
  3. Progress Feedback:
    • A QProgressBar is updated via a signal from the worker thread, providing visual feedback to the user that an operation is in progress, improving the perceived performance and user experience.
  4. Resource Management:
    • Explicitly managing QThread and QObject lifecycles (moving the worker to the thread, connecting finished signals for cleanup) prevents resource leaks and ensures graceful shutdown of background operations.

Reference