How to Safely Load Shapefiles into QgsProject Without UI Blocking

"""
safe_shapefile_loader.py — Non-blocking shapefile loader for QgsProject.

Drop this module into your plugin's lib/ directory and call submit_loader()
from any UI action. Compatible with QGIS 3.28+.
"""

import os
from typing import Optional

from qgis.core import (
    QgsApplication,
    QgsCoordinateReferenceSystem,
    QgsMessageLog,
    QgsProject,
    QgsTask,
    QgsVectorLayer,
    Qgis,
)
from qgis.PyQt.QtCore import pyqtSignal


class SafeShapefileLoader(QgsTask):
    """Background task that parses a shapefile off the main thread.

    Emits layer_loaded(layer_name) on success or task_failed(error_message)
    on failure. Layer registration happens inside finished(), which Qt
    guarantees executes on the GUI thread.
    """

    layer_loaded = pyqtSignal(str)   # emitted with the registered layer name
    task_failed = pyqtSignal(str)    # emitted with a human-readable error

    def __init__(
        self,
        shapefile_path: str,
        layer_name: Optional[str] = None,
        fallback_crs: Optional[str] = None,
    ) -> None:
        """
        Args:
            shapefile_path: Absolute path to the .shp file.
            layer_name:     Display name in the Layers panel.
                            Defaults to the filename stem.
            fallback_crs:   EPSG string (e.g. 'EPSG:4326') applied when the
                            .prj sidecar is missing, skipping a live registry
                            lookup. Pass None to let QGIS resolve automatically.
        """
        stem = os.path.splitext(os.path.basename(shapefile_path))[0]
        super().__init__(
            f"Loading {stem}",
            QgsTask.CanCancel,
        )
        self.shapefile_path = shapefile_path
        self.layer_name = layer_name or stem
        self.fallback_crs = fallback_crs

        self._layer: Optional[QgsVectorLayer] = None
        self._error: Optional[str] = None

    # ------------------------------------------------------------------
    # Background thread — no GUI operations allowed here
    # ------------------------------------------------------------------

    def run(self) -> bool:
        """Parse the shapefile.  Executes in a QThreadPool worker thread.

        Returns True on success so that finished(result=True) is called on the
        main thread. Returns False on validation failure or cancellation.
        """
        if self.isCanceled():
            return False

        try:
            # Heavy I/O: OGR reads .shp/.shx/.dbf, resolves geometry types
            layer = QgsVectorLayer(self.shapefile_path, self.layer_name, "ogr")

            if not layer.isValid():
                provider_error = layer.dataProvider().error().message()
                self._error = provider_error or f"Invalid layer: {self.shapefile_path}"
                return False

            # Apply fallback CRS before registration to avoid a live EPSG query
            if self.fallback_crs and not layer.crs().isValid():
                layer.setCrs(QgsCoordinateReferenceSystem(self.fallback_crs))

            # Pre-compute extents so the first render is immediate
            if layer.isSpatial():
                layer.updateExtents()

            self._layer = layer
            return True

        except Exception as exc:  # noqa: BLE001
            self._error = str(exc)
            return False

    # ------------------------------------------------------------------
    # Main GUI thread — safe to touch QgsProject and signals here
    # ------------------------------------------------------------------

    def finished(self, result: bool) -> None:  # type: ignore[override]
        """Register the layer. Qt marshals this call back to the GUI thread.

        Args:
            result: True if run() returned True; False on failure or cancel.
        """
        if result and self._layer is not None:
            QgsProject.instance().addMapLayer(self._layer)
            self.layer_loaded.emit(self.layer_name)
            QgsMessageLog.logMessage(
                f"Layer loaded: {self.layer_name}",
                "SafeShapefileLoader",
                Qgis.Success,
            )
        else:
            message = self._error or "Task cancelled"
            self.task_failed.emit(message)
            QgsMessageLog.logMessage(
                f"Load failed: {message}",
                "SafeShapefileLoader",
                Qgis.Warning,
            )


def submit_loader(
    shapefile_path: str,
    layer_name: Optional[str] = None,
    fallback_crs: Optional[str] = None,
) -> SafeShapefileLoader:
    """Convenience factory: build the task, wire default logging, and submit.

    Returns the task instance so callers can connect additional signals.
    """
    task = SafeShapefileLoader(shapefile_path, layer_name, fallback_crs)
    QgsApplication.taskManager().addTask(task)
    return task

from qgis.PyQt.QtWidgets import QAction, QFileDialog
from qgis.core import QgsMessageLog, Qgis
from .safe_shapefile_loader import submit_loader


class MyPlugin:
    """Minimal plugin skeleton showing loader integration."""

    def __init__(self, iface):
        self.iface = iface

    def initGui(self) -> None:
        self.action = QAction("Load Shapefile (non-blocking)", self.iface.mainWindow())
        self.action.triggered.connect(self._on_load)
        self.iface.addToolBarIcon(self.action)

    def _on_load(self) -> None:
        path, _ = QFileDialog.getOpenFileName(
            self.iface.mainWindow(), "Select Shapefile", "", "Shapefiles (*.shp)"
        )
        if not path:
            return

        task = submit_loader(path, fallback_crs="EPSG:4326")
        task.task_failed.connect(
            lambda err: self.iface.messageBar().pushWarning("Loader", err)
        )
        task.layer_loaded.connect(
            lambda name: self.iface.messageBar().pushSuccess("Loader", f"Loaded: {name}")
        )

    def unload(self) -> None:
        self.iface.removeToolBarIcon(self.action)
        del self.action

"""
headless_load.py — load a shapefile in a QgsApplication (no UI) context.

Run with:
    python headless_load.py /data/boundaries.shp
"""

import sys
from qgis.core import QgsApplication
from safe_shapefile_loader import submit_loader


def main(shapefile_path: str) -> int:
    app = QgsApplication([], False)  # False = no GUI
    app.initQgis()

    finished_flag = [False]

    task = submit_loader(shapefile_path)
    task.layer_loaded.connect(lambda _: finished_flag.__setitem__(0, True))
    task.task_failed.connect(lambda err: (print(f"Error: {err}"), finished_flag.__setitem__(0, True)))

    # Pump the event loop until the task completes
    while not finished_flag[0]:
        app.processEvents()

    app.exitQgis()
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1]))