Asynchronous Task Execution with QgsTask

from qgis.core import QgsTask, QgsMessageLog, Qgis, QgsGeometry
from qgis.PyQt.QtCore import pyqtSignal
from typing import Optional


class HeavyVectorProcessor(QgsTask):
    """Background task that processes pre-fetched geometries without blocking the UI.

    All geometry data must be pre-fetched and cloned on the main thread
    before this task is submitted to the task manager.
    """

    # Custom signal for thread-safe progress reporting — Qt marshals
    # arguments across the thread boundary automatically.
    progress_updated = pyqtSignal(int, str)

    def __init__(self, geometries: list[QgsGeometry], buffer_distance: float) -> None:
        """
        Parameters
        ----------
        geometries : list[QgsGeometry]
            Pre-fetched and cloned on the main thread before task submission.
        buffer_distance : float
            Buffer distance in layer CRS units.
        """
        super().__init__(
            f"Processing {len(geometries)} geometries",
            QgsTask.CanCancel,
        )
        # Clone each geometry so this task owns its own copies independently
        # of any QgsVectorLayer that might be edited while the task runs.
        self.geometries: list[QgsGeometry] = [g.clone() for g in geometries]
        self.buffer_distance = buffer_distance
        self.processed_count: int = 0
        self.result_geometries: list[QgsGeometry] = []
        self.error_msg: Optional[str] = None

    def run(self) -> bool:
        """Execute in a background thread. Return True on success, False on failure.

        Never access QgsProject, iface, or any Qt widget from this method.
        All data exchange with the main thread goes through instance variables
        read in finished(), or via queued signals.
        """
        try:
            total = len(self.geometries)
            if total == 0:
                self.error_msg = "No geometries to process."
                return False

            for i, geom in enumerate(self.geometries):
                # Check cancellation at the start of each iteration
                if self.isCanceled():
                    return False

                # Geometry operations are safe in a worker thread — they do
                # not touch the Qt object tree or any QGIS project state.
                buffered = geom.buffer(self.buffer_distance, 8)
                self.result_geometries.append(buffered)
                self.processed_count += 1

                progress_pct = int((i + 1) / total * 100)
                self.setProgress(progress_pct)
                self.progress_updated.emit(progress_pct, f"Processing {i + 1}/{total}")

            return True

        except Exception as exc:
            self.error_msg = str(exc)
            QgsMessageLog.logMessage(
                f"Task failed: {exc}",
                "HeavyVectorProcessor",
                level=Qgis.Critical,
            )
            return False

    def finished(self, result: bool) -> None:
        """Execute on the main thread after run() completes.

        Safe to call iface, QgsProject.instance().addMapLayer(), or any
        Qt widget method from here.
        """
        if result:
            QgsMessageLog.logMessage(
                f"Completed — processed {self.processed_count} geometries.",
                "HeavyVectorProcessor",
                level=Qgis.Success,
            )
            # Add results to canvas, refresh symbology, or open a results dialog
        elif self.isCanceled():
            QgsMessageLog.logMessage(
                "Task cancelled by user.",
                "HeavyVectorProcessor",
                level=Qgis.Warning,
            )
        else:
            QgsMessageLog.logMessage(
                f"Task failed: {self.error_msg}",
                "HeavyVectorProcessor",
                level=Qgis.Critical,
            )

from qgis.core import (
    QgsApplication,
    QgsProject,
    QgsMessageLog,
    Qgis,
)
from qgis.PyQt.QtCore import Qt


def execute_background_task(layer_id: str, distance: float) -> None:
    """Create and submit a HeavyVectorProcessor task.

    Parameters
    ----------
    layer_id : str
        The map layer ID as returned by QgsMapLayer.id().
    distance : float
        Buffer distance in layer CRS units.
    """
    # --- Main thread: read all data before task submission ---
    layer = QgsProject.instance().mapLayer(layer_id)
    if not layer or not layer.isValid():
        QgsMessageLog.logMessage(
            "Layer not found or invalid.",
            "HeavyVectorProcessor",
            level=Qgis.Warning,
        )
        return

    # Iterate features and clone geometries on the main thread.
    # Do NOT pass the QgsVectorLayer itself — the task must not
    # call layer methods from the worker thread.
    geometries = [f.geometry() for f in layer.getFeatures()]

    # --- Create and wire the task ---
    task = HeavyVectorProcessor(geometries, distance)

    # Connect signals before addTask() — the task may start immediately.
    task.progress_updated.connect(
        lambda pct, msg: update_progress_bar(pct, msg),
        Qt.QueuedConnection,
    )
    task.taskCompleted.connect(lambda: handle_success(task))
    task.taskTerminated.connect(lambda: handle_failure(task))

    QgsApplication.taskManager().addTask(task)
    QgsMessageLog.logMessage(
        f"Submitted task for {len(geometries)} geometries.",
        "HeavyVectorProcessor",
    )

from qgis.core import QgsVectorLayer, QgsProject, QgsWkbTypes, QgsFeature
from qgis.PyQt.QtWidgets import QMessageBox


def update_progress_bar(percent: int, message: str) -> None:
    """Main-thread slot for per-iteration progress updates.

    Throttle to ~10-20 Hz if the dataset is very large; emitting
    signals at thousands of Hz will saturate the event queue.
    """
    # iface.mainWindow().statusBar().showMessage(f"{message} ({percent}%)")
    pass


def handle_success(task: HeavyVectorProcessor) -> None:
    """Build a memory layer from task results and add it to the project."""
    result_layer = QgsVectorLayer(
        "Polygon?crs=EPSG:4326",
        "Buffered Geometries",
        "memory",
    )
    result_layer.startEditing()
    for geom in task.result_geometries:
        feat = QgsFeature()
        feat.setGeometry(geom)
        result_layer.addFeature(feat)
    result_layer.commitChanges()
    QgsProject.instance().addMapLayer(result_layer)


def handle_failure(task: HeavyVectorProcessor) -> None:
    """Notify the user when a task ends without producing results."""
    if task.isCanceled():
        return  # User-initiated; no error dialog needed
    QMessageBox.critical(
        None,
        "Processing error",
        f"Task failed: {task.error_msg}",
    )

from qgis.core import QgsTask, QgsApplication
from typing import Optional


class ReprojectionTask(QgsTask):
    """Reproject geometries to a target CRS in a worker thread."""

    def __init__(self, geometries: list, target_crs_auth_id: str) -> None:
        super().__init__("Reprojecting features", QgsTask.CanCancel)
        self.geometries = [g.clone() for g in geometries]
        self.target_crs_auth_id = target_crs_auth_id
        self.reprojected: list = []

    def run(self) -> bool:
        from qgis.core import QgsCoordinateTransform, QgsCoordinateReferenceSystem, QgsProject
        # Note: QgsCoordinateTransform is safe to construct in a worker thread
        # because it reads CRS definitions from files, not from QgsProject state.
        target_crs = QgsCoordinateReferenceSystem(self.target_crs_auth_id)
        for geom in self.geometries:
            if self.isCanceled():
                return False
            self.reprojected.append(geom.clone())
        return True

    def finished(self, result: bool) -> None:
        if result:
            # Coordinate transformations and CRS handling are covered in
            # /pyqgis-core-architecture-data-handling/coordinate-transformations-and-crs-handling/
            pass


class SpatialJoinTask(QgsTask):
    """Perform a spatial join once reprojection completes."""

    def __init__(self, reproject_task: ReprojectionTask) -> None:
        super().__init__("Spatial join", QgsTask.CanCancel)
        # Store a reference so finished() can read reprojected geometries
        self.reproject_task = reproject_task

    def run(self) -> bool:
        geometries = self.reproject_task.reprojected
        # Heavy spatial join logic here …
        return True

    def finished(self, result: bool) -> None:
        pass


def run_pipeline(geometries: list) -> None:
    """Submit a two-stage reprojection → join pipeline."""
    reproject = ReprojectionTask(geometries, "EPSG:32632")
    join = SpatialJoinTask(reproject)

    # addSubTask ensures join only starts after reproject completes
    reproject.addSubTask(join, [], QgsTask.ParentDependsOnSubTask)
    QgsApplication.taskManager().addTask(reproject)

from qgis.core import QgsApplication, QgsTask
from typing import Optional


def _compute_area_sum(task: QgsTask, geometries: list) -> Optional[dict]:
    """Compute the total area of a geometry list in a worker thread.

    Parameters
    ----------
    task : QgsTask
        The hosting task object — use task.isCanceled() to check for cancellation.
    geometries : list[QgsGeometry]
        Pre-cloned geometries passed in from the main thread.
    """
    total = 0.0
    for i, geom in enumerate(geometries):
        if task.isCanceled():
            return None
        total += geom.area()
        task.setProgress(int((i + 1) / len(geometries) * 100))
    return {"total_area": total, "count": len(geometries)}


def _on_area_complete(exception: Optional[Exception], result: Optional[dict] = None) -> None:
    """Receive the result dict (or exception) on the main thread."""
    if exception:
        raise exception
    if result:
        print(f"Total area: {result['total_area']:.2f}, features: {result['count']}")


# geometries must be pre-fetched and cloned on the main thread
task = QgsTask.fromFunction(
    "Compute area sum",
    _compute_area_sum,
    on_finished=_on_area_complete,
    geometries=pre_fetched_geometries,
)
QgsApplication.taskManager().addTask(task)