PyQGIS Engineering & Automation

Creating a Reusable PyQGIS Processing Algorithm Template

Step-by-step guide to building a production-ready, reusable QgsProcessingAlgorithm subclass in QGIS 3.x — covering parameter definition, safe feature…

Subclass QgsProcessingAlgorithm, implement the mandatory metadata and I/O methods, register the class with a QgsProcessingProvider, and route all progress and cancellation signals through QgsProcessingFeedback. This is part of the Building Custom Processing Algorithms guide, which covers the full lifecycle from parameter chaining to graphical model integration.

Complete Runnable Template

The code below is a drop-in, QGIS 3.x-compatible template. Replace the business logic comment with your own spatial or attribute operations; every other piece — parameter resolution, sink creation, feature iteration, progress tracking — is production-ready as written.

python
from __future__ import annotations

from qgis.core import (
    QgsProcessingAlgorithm,
    QgsProcessingContext,
    QgsProcessingException,
    QgsProcessingFeedback,
    QgsProcessingParameterFeatureSource,
    QgsProcessingParameterNumber,
    QgsProcessingParameterVectorDestination,
    QgsProcessing,
    QgsFeature,
    QgsFeatureSink,
)


class ReusableTemplateAlgorithm(QgsProcessingAlgorithm):
    """
    Production-ready template for a custom PyQGIS processing algorithm.

    Drop this into a QgsProcessingProvider subclass and extend
    processAlgorithm() with your domain logic. See the registration
    snippet below for how to wire it into a plugin provider.
    """

    # ------------------------------------------------------------------ #
    # Parameter keys — define as class constants to avoid typos           #
    # ------------------------------------------------------------------ #
    INPUT: str = "INPUT"
    OUTPUT: str = "OUTPUT"
    THRESHOLD: str = "THRESHOLD"

    # ------------------------------------------------------------------ #
    # Metadata                                                             #
    # ------------------------------------------------------------------ #

    def name(self) -> str:
        """Unique, lowercase identifier used by processing.run()."""
        return "reusable_template"

    def displayName(self) -> str:
        """Human-readable name shown in the Processing Toolbox."""
        return "Reusable Processing Template"

    def group(self) -> str:
        """Toolbox group label (localisable)."""
        return "Custom Automation"

    def groupId(self) -> str:
        """Stable lowercase group identifier."""
        return "custom_automation"

    def shortHelpString(self) -> str:
        """Shown in the algorithm's Help panel inside QGIS."""
        return (
            "A production-ready template for building custom PyQGIS processing "
            "algorithms. Extend processAlgorithm() with your spatial logic."
        )

    def createInstance(self) -> "ReusableTemplateAlgorithm":
        """
        Called by the framework to spawn a fresh instance per execution.
        Never share state via class attributes — createInstance() is the
        boundary that prevents state leakage between runs.
        """
        return ReusableTemplateAlgorithm()

    # ------------------------------------------------------------------ #
    # Parameter / output declaration                                       #
    # ------------------------------------------------------------------ #

    def initAlgorithm(self, config: dict | None = None) -> None:
        """
        Declare all inputs and outputs.  The Processing Framework renders
        these as widgets in the dialog, batch processor, and modeler
        automatically — no UI code required here.
        """
        self.addParameter(
            QgsProcessingParameterFeatureSource(
                self.INPUT,
                "Input vector layer",
                types=[QgsProcessing.TypeVectorAnyGeometry],
            )
        )
        self.addParameter(
            QgsProcessingParameterNumber(
                self.THRESHOLD,
                "Processing threshold",
                type=QgsProcessingParameterNumber.Double,
                defaultValue=0.5,
                minValue=0.0,
                maxValue=1.0,
                optional=False,
            )
        )
        self.addParameter(
            QgsProcessingParameterVectorDestination(
                self.OUTPUT,
                "Output vector layer",
            )
        )

    # ------------------------------------------------------------------ #
    # Execution                                                            #
    # ------------------------------------------------------------------ #

    def processAlgorithm(
        self,
        parameters: dict,
        context: QgsProcessingContext,
        feedback: QgsProcessingFeedback,
    ) -> dict:
        """
        Core worker method.  Runs on the main thread inside QGIS Desktop
        and on the calling thread in headless scripts.

        Returns a dict mapping output parameter keys to their resolved IDs.
        Raise QgsProcessingException for any unrecoverable error — the
        framework converts this into a user-visible error message and
        cleans up temporary files.
        """
        # 1. Resolve and validate input source
        source = self.parameterAsSource(parameters, self.INPUT, context)
        if source is None:
            raise QgsProcessingException(
                self.invalidSourceError(parameters, self.INPUT)
            )

        # 2. Resolve scalar parameters
        threshold: float = self.parameterAsDouble(
            parameters, self.THRESHOLD, context
        )

        # 3. Create output sink — inherits fields, geometry type, and CRS
        #    from the source so downstream tools receive a compatible layer.
        sink, dest_id = self.parameterAsSink(
            parameters,
            self.OUTPUT,
            context,
            source.fields(),
            source.wkbType(),
            source.sourceCrs(),
        )
        if sink is None:
            raise QgsProcessingException(
                self.invalidSinkError(parameters, self.OUTPUT)
            )

        # 4. Iterate features with progress reporting and cancellation support
        feature_count: int = source.featureCount()
        step: float = 100.0 / feature_count if feature_count else 0.0
        processed: int = 0

        for current, feature in enumerate(source.getFeatures()):
            if feedback.isCanceled():
                feedback.pushWarning("Processing cancelled by user.")
                break

            # ---- Replace this block with your domain logic ---- #
            # Example: pass-through, but threshold could filter or
            # transform attributes before writing to the sink.
            if threshold >= 0.0:
                out_feature = QgsFeature(source.fields())
                out_feature.setGeometry(feature.geometry())
                out_feature.setAttributes(feature.attributes())
                # FastInsert skips redundant geometry validation — up to 50%
                # faster for bulk writes.  Use AddFeatures if you need
                # feature-level error signals during debugging.
                sink.addFeature(out_feature, QgsFeatureSink.FastInsert)
                processed += 1
            # ---- End domain logic ---- #

            feedback.setProgress(int(current * step))

        feedback.pushInfo(f"Processed {processed} of {feature_count} features.")
        return {self.OUTPUT: dest_id}

Execution Flow Diagram

The diagram below traces the algorithm lifecycle from the moment the Processing Framework calls processAlgorithm() through feature iteration to the returned result dictionary.

PyQGIS processing algorithm execution flow Flowchart showing the sequence: processAlgorithm is called, then parameterAsSource validates input, then parameterAsSink creates output, then a feature iteration loop checks feedback.isCanceled — if cancelled it breaks to the result, if not it runs domain logic and addFeature, loops back, and when features are exhausted returns the result dictionary. processAlgorithm() parameterAsSource() — validate parameterAsSink() — create output Next feature? (getFeatures loop) feedback.isCanceled()? domain logic sink.addFeature() return {OUTPUT: dest_id} no more yes yes → break no

Architecture Breakdown

Metadata and createInstance()

name() returns the stable lowercase identifier that processing.run() uses to locate the algorithm. displayName(), group(), and groupId() control how the algorithm appears in the Processing Toolbox’s search index and UI tree. shortHelpString() populates the in-dialog Help panel.

createInstance() is the critical lifecycle boundary. The Processing Framework calls it to produce a fresh object for each execution. Because QGIS can batch-run algorithms and queue them for the graphical modeler, you must never store mutable execution state in class attributes — all per-run data must live in local variables inside processAlgorithm().

initAlgorithm() — Parameter Contract

initAlgorithm() declares the algorithm’s public interface. The Processing Framework reads this declaration to render the dialog widgets, construct the modeler port, and validate inputs before calling processAlgorithm(). Using QgsProcessingParameterFeatureSource (rather than a plain layer parameter) accepts layers, memory layers, and outputs piped from upstream algorithm steps in a model — it is the correct choice for any vector input. QgsProcessingParameterVectorDestination similarly handles file-backed, memory-backed, and GeoPackage outputs transparently.

Define parameter keys as class constants (INPUT = "INPUT"). String literals scattered across initAlgorithm() and processAlgorithm() are a common source of silent failures that only surface at runtime.

processAlgorithm() — Execution Contract

processAlgorithm() receives three arguments: parameters (a dict of resolved parameter values), context (a QgsProcessingContext carrying the project reference, temporary directory, and CRS transform settings), and feedback (a QgsProcessingFeedback instance that bridges to the progress bar and log panel).

The method contract has four non-negotiable invariants:

  1. Validate every source and sink. parameterAsSource() returns None when the referenced layer is invalid or missing. Raise QgsProcessingException with self.invalidSourceError() immediately — do not proceed to the feature loop with a None source.
  2. Check feedback.isCanceled() every iteration. Without this check the algorithm is unresponsive to the Cancel button and blocks the QGIS event loop during long-running operations.
  3. Call feedback.setProgress() with a 0–100 float. Pre-compute step = 100.0 / feature_count before the loop rather than dividing inside it.
  4. Return a dict mapping every output key to its resolved ID. Omitting an output key from the return dict causes downstream model steps that reference that output to silently receive None.

Output Sink Geometry and Field Inheritance

Passing source.fields(), source.wkbType(), and source.sourceCrs() to parameterAsSink() ensures the output sink inherits the input schema exactly. This matters for coordinate transformations — if your algorithm needs to reproject features, apply a QgsCoordinateTransform inside the loop and pass the target CRS to parameterAsSink() instead of source.sourceCrs().

Registration and Integration

Plugin Provider Registration

To expose the algorithm in the Processing Toolbox, register it through a QgsProcessingProvider subclass. This is the standard approach for plugin lifecycle management — the provider is added during initGui() and removed in unload().

python
from __future__ import annotations

from qgis.core import QgsApplication, QgsProcessingProvider

from .reusable_template import ReusableTemplateAlgorithm


class CustomAutomationProvider(QgsProcessingProvider):
    """Registers all custom algorithms for this plugin."""

    def id(self) -> str:
        return "custom_automation"

    def name(self) -> str:
        return "Custom Automation"

    def loadAlgorithms(self) -> None:
        """Called by the framework when the provider is activated."""
        self.addAlgorithm(ReusableTemplateAlgorithm())
        # Add further algorithm instances here as the library grows.


# In your plugin class __init__.py:
class MyPlugin:
    def __init__(self, iface):
        self._provider = CustomAutomationProvider()

    def initGui(self) -> None:
        QgsApplication.processingRegistry().addProvider(self._provider)

    def unload(self) -> None:
        QgsApplication.processingRegistry().removeProvider(self._provider)

Standalone Headless Script

For CI pipelines or server-side batch jobs that run outside QGIS Desktop, initialise the QGIS application context before calling processing.run():

python
from __future__ import annotations

import sys
from qgis.core import QgsApplication

# Adjust the prefix path to your QGIS installation.
QgsApplication.setPrefixPath("/usr", True)
qgs = QgsApplication([], False)
qgs.initQgis()

# Import processing after initQgis() so the framework is ready.
from qgis import processing
from qgis.core import QgsProcessingFeedback

feedback = QgsProcessingFeedback()
result = processing.run(
    "custom_automation:reusable_template",
    {
        "INPUT": "/data/input.gpkg|layername=parcels",
        "THRESHOLD": 0.75,
        "OUTPUT": "TEMPORARY_OUTPUT",
    },
    feedback=feedback,
)
print(result["OUTPUT"])

qgs.exitQgis()

Note that processing.run() resolves the algorithm by the <provider_id>:<algorithm_name> string returned by id() and name() respectively. Registering the provider before the call is mandatory — the framework will raise a QgsProcessingException if the algorithm ID is unknown.

Production Best Practices

  • Never mutate input layers in place. Write exclusively to the output sink. The framework assumes processAlgorithm() is side-effect-free so it can safely replay executions in the modeler.
  • Use QgsFeatureSink.FastInsert for bulk writes. It bypasses per-feature geometry validation and accelerates large writes by 30–50%. Switch to QgsFeatureSink.AddFeatures only when you need feature-level insert error signals during local debugging.
  • Keep algorithms stateless. Store configuration in QgsProcessingContext extras or project variables, not in class attributes. Stateful algorithms fail silently when the framework spawns multiple instances for batch or parallel model execution.
  • Log with feedback.pushInfo() and feedback.pushWarning(). These route to the algorithm’s log panel in the Processing dialog. print() goes to the Python console only and is invisible when the algorithm runs headlessly or inside the modeler.
  • Handle CRS transformations explicitly. If your algorithm requires a fixed projection, accept the input CRS via QgsProcessingParameterFeatureSource and apply a QgsCoordinateTransform inside the loop rather than assuming the input layer’s CRS matches your target. See the coordinate transformations and CRS handling guide for the correct transform chain.
  • Test with a mock QgsProcessingFeedback. Subclass or mock QgsProcessingFeedback in your test suite to assert that setProgress() increases monotonically and that isCanceled() triggers the break path. The QGIS PyQGIS API reference documents all method signatures you need to implement for a compliant mock.
  • Use type hints throughout. parameterAsSource(), parameterAsSink(), and parameterAsDouble() all return typed values in QGIS 3.28+. Type hints make contract violations visible at static analysis time and reduce runtime AttributeError surprises.

Related