Building Custom Processing Algorithms in PyQGIS

The QGIS Processing Framework provides a standardized, execution-agnostic pipeline for geospatial operations. While the built-in algorithm library covers extensive vector, raster, and database workflows, enterprise GIS projects frequently require domain-specific logic that falls outside standard toolboxes. Building custom processing algorithms lets development teams encapsulate proprietary spatial routines, enforce organizational data standards, and expose reusable logic directly within the QGIS Processing Toolbox, Python console, and graphical modeler. This page is part of the Plugin Development & UI Integration guide and assumes you are already familiar with the broader plugin lifecycle and resource management contract.

Prerequisites #

Before implementing custom algorithms, ensure your environment meets these baseline requirements:

• QGIS 3.28 LTR or 3.34+: The Processing Framework API stabilized in the 3.x series. Earlier versions lack modern parameter validation, sink management, and consistent thread-safety guarantees.
• Python 3.9+: QGIS ships with a bundled Python interpreter. Use qgis_process for headless execution testing or the built-in Python Console for interactive debugging.
• PyQGIS core fluency: Understanding QgsVectorLayer, QgsFeature, QgsGeometry, and coordinate reference system handling is mandatory. The Processing API relies on these classes for all data I/O.
• Processing Framework concepts: Algorithms must inherit from QgsProcessingAlgorithm, implement required abstract methods, and register through a QgsProcessingProvider. The framework manages execution context, threading, and modeler integration automatically.
• Validation dataset: Prepare a small vector layer (< 500 features) with mixed geometry types to smoke-test parameter resolution and output sinks before connecting to production data.

Architecture and Internals #

The Processing Framework separates algorithm definition from execution orchestration. Each custom tool inherits from QgsProcessingAlgorithm, which specifies the contract for inputs, outputs, and the execution body. The framework wraps these definitions in a QgsProcessingProvider, which acts as a registry container. When QGIS starts — or when your plugin calls addProvider() — the provider loads its algorithms into the global QgsApplication.processingRegistry(). This architecture decouples the same algorithm body from its execution surface: the identical code runs whether triggered from the GUI dialog, the graphical modeler, a Python script, or the headless qgis_process CLI.

The C++ side manages memory ownership: QgsProcessingContext holds a temporary layer store, and the framework transfers output layer ownership from that store back to QgsProject after execution. Understanding this ownership model prevents the most common class of memory leak in custom algorithms — holding references to QgsVectorLayer objects that the context has already destroyed.

Thread safety is a first-class concern. The framework may schedule processAlgorithm() on a background thread via QgsTask. Never interact with the QGIS main window, map canvas, or any UI widget from inside the execution body. Use QgsProcessingFeedback for progress and status messages, and QgsProcessingException for hard failures.

Step-by-Step Implementation #

1. Define Algorithm Metadata #

Every algorithm requires a unique identifier, display name, group classification, and short description. The name() return value is used as the stable algorithm ID in scripts, modeler files, and qgis_process calls — it must never change once published. displayName() is the human-readable label shown in the Processing Toolbox. The group() and groupId() pair determines the collapsible category. shortHelpString() provides inline documentation rendered in the algorithm dialog’s help panel.

You must also implement createInstance(), which returns a fresh instance of the algorithm class. QGIS instantiates algorithms dynamically during each execution rather than reusing a single object, so this factory method is not optional.

from qgis.core import QgsProcessingAlgorithm


class BufferAndFilterAlgorithm(QgsProcessingAlgorithm):
    """Buffer input features and filter by a numeric attribute threshold.

    Registered under the 'Enterprise Workflows' group. Algorithm ID:
    enterprise_workflows:custombufferfilter
    """

    def name(self) -> str:
        return "custombufferfilter"

    def displayName(self) -> str:
        return "Buffer and Filter"

    def group(self) -> str:
        return "Enterprise Workflows"

    def groupId(self) -> str:
        return "enterprise_workflows"

    def shortHelpString(self) -> str:
        return (
            "Applies a configurable buffer distance to all input features, "
            "then filters the result by a numeric attribute threshold. "
            "Input and output CRS must match — transform beforehand if needed."
        )

    def createInstance(self) -> "BufferAndFilterAlgorithm":
        return BufferAndFilterAlgorithm()

2. Configure Parameters and Outputs #

Override initAlgorithm() to declare inputs and outputs using QgsProcessingParameter subclasses. Parameters define the contract between the UI, the modeler, and the execution engine. The framework serializes these definitions to JSON when saving modeler files and to command-line flags when calling qgis_process.

Use QgsProcessingParameterFeatureDestination for every vector output rather than a hardcoded path. This allows QGIS to manage temporary memory layers, handle batch outputs, and respect the user’s default storage location — all without any extra code in your algorithm.

from qgis.core import (
    QgsProcessingAlgorithm,
    QgsProcessingParameterFeatureSource,
    QgsProcessingParameterFeatureDestination,
    QgsProcessingParameterNumber,
    QgsProcessingParameterField,
    QgsProcessing,
)


def initAlgorithm(self, config: dict | None = None) -> None:
    """Declare all input parameters and output sinks."""
    self.addParameter(
        QgsProcessingParameterFeatureSource(
            "INPUT",
            "Input vector layer",
            [QgsProcessing.TypeVectorAnyGeometry],
        )
    )
    self.addParameter(
        QgsProcessingParameterNumber(
            "BUFFER_DIST",
            "Buffer distance (map units)",
            type=QgsProcessingParameterNumber.Double,
            defaultValue=100.0,
            minValue=0.0,
        )
    )
    self.addParameter(
        QgsProcessingParameterField(
            "FILTER_FIELD",
            "Numeric field for threshold filter",
            parentLayerParameterName="INPUT",
            type=QgsProcessingParameterField.Numeric,
            optional=True,
        )
    )
    self.addParameter(
        QgsProcessingParameterNumber(
            "THRESHOLD",
            "Minimum field value to keep",
            type=QgsProcessingParameterNumber.Double,
            defaultValue=0.0,
            optional=True,
        )
    )
    self.addParameter(
        QgsProcessingParameterFeatureDestination(
            "OUTPUT",
            "Buffered output layer",
        )
    )

When your algorithm requires complex user input beyond standard parameters — for example, a multi-column selection table or an interactive map click — the designing Qt dialogs and form widgets guide covers how to build custom parameter widgets that integrate seamlessly with the Processing dialog.

3. Implement Core Processing Logic #

processAlgorithm() is the execution body. It receives a dictionary of resolved parameters, a QgsProcessingContext that holds the temporary layer store and project reference, and a QgsProcessingFeedback for progress reporting and cancellation. All heavy computation belongs here and only here.

Validate every resolved parameter immediately and raise QgsProcessingException on failure — never silently skip bad inputs, because the framework treats None returns from parameter resolution as user configuration errors.

from qgis.core import (
    QgsFeature,
    QgsFeatureSink,
    QgsProcessingContext,
    QgsProcessingException,
    QgsProcessingFeedback,
)


def processAlgorithm(
    self,
    parameters: dict,
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict:
    """Execute the buffer-and-filter operation.

    Returns a dict mapping output parameter names to destination IDs.
    The framework uses these IDs to transfer layers from the context's
    temporary store into QgsProject after execution completes.
    """
    source = self.parameterAsSource(parameters, "INPUT", context)
    if source is None:
        raise QgsProcessingException(
            self.invalidSourceError(parameters, "INPUT")
        )

    buffer_dist: float = self.parameterAsDouble(parameters, "BUFFER_DIST", context)
    filter_field: str = self.parameterAsString(parameters, "FILTER_FIELD", context)
    threshold: float = self.parameterAsDouble(parameters, "THRESHOLD", context)
    use_filter: bool = bool(filter_field)

    # Open the output sink — geometry type may change after buffering (points→polygons)
    (sink, dest_id) = self.parameterAsSink(
        parameters,
        "OUTPUT",
        context,
        source.fields(),
        source.wkbType(),   # buffer() preserves geometry family for polygon input
        source.sourceCrs(),
    )
    if sink is None:
        raise QgsProcessingException(
            self.invalidSinkError(parameters, "OUTPUT")
        )

    feature_count = source.featureCount()
    total = 100.0 / feature_count if feature_count else 0.0
    field_idx = source.fields().lookupField(filter_field) if use_filter else -1

    for current, feature in enumerate(source.getFeatures()):
        # Honour cancellation at every iteration — critical for long-running jobs
        if feedback.isCanceled():
            break

        # Attribute-based pre-filter before the more expensive geometry operation
        if use_filter and field_idx >= 0:
            value = feature.attribute(field_idx)
            if value is None or float(value) < threshold:
                feedback.setProgress(int(current * total))
                continue

        buffered_geom = feature.geometry().buffer(buffer_dist, segments=16)
        out_feature = QgsFeature(feature)
        out_feature.setGeometry(buffered_geom)
        # FastInsert skips duplicate-ID checks — safe when building a new sink
        sink.addFeature(out_feature, QgsFeatureSink.FastInsert)
        feedback.setProgress(int(current * total))

    return {"OUTPUT": dest_id}

For vector and raster data access patterns beyond simple getFeatures() iteration — such as spatial filtering, attribute expression filters, or raster block reading — the linked guide covers request objects and optimized access methods that significantly reduce I/O overhead on large datasets.

4. Register with a Custom Provider #

Custom algorithms must be attached to a QgsProcessingProvider. The provider exposes all contained algorithms to the global registry during startup and again on plugin reload. Registration typically occurs in your plugin’s initGui() method; deregistration must happen in unload() to prevent stale algorithm references after the plugin is disabled.

from qgis.core import QgsApplication, QgsProcessingProvider


class EnterpriseProvider(QgsProcessingProvider):
    """Processing provider for enterprise spatial utility algorithms.

    Register via QgsApplication.processingRegistry().addProvider(provider)
    in your plugin's initGui(). Remove in unload() to prevent leaks.
    See: /plugin-development-ui-integration/plugin-lifecycle-and-resource-management/
    """

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

    def name(self) -> str:
        return "Enterprise Spatial Tools"

    def longName(self) -> str:
        return "Enterprise Spatial Tools (internal)"

    def loadAlgorithms(self) -> None:
        """Called by the framework whenever the registry rebuilds.

        Add every algorithm class here. The framework calls createInstance()
        on each one — never pass live instances shared across calls.
        """
        self.addAlgorithm(BufferAndFilterAlgorithm())

    def icon(self):
        return QgsApplication.getThemeIcon("/mIconProcessingAlgorithm.svg")


# --- Plugin class (excerpt) ---
class EnterprisePlugin:
    def __init__(self, iface):
        self.iface = iface
        self._provider: EnterpriseProvider | None = None

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

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

Once registered, your algorithms appear in the Processing Toolbox and are immediately available to the graphical modeler and qgis_process. To bind them to toolbar buttons or menu actions, use processing.execAlgorithmDialog() or processing.run() from the processing module rather than invoking processAlgorithm() directly — this keeps the execution pathway consistent and preserves feedback routing.

Advanced Patterns #

Composing Algorithms with processing.run() #

Custom algorithms can invoke other registered algorithms from within their own processAlgorithm() body, enabling modular, composable pipelines without duplicating spatial logic. Pass the parent context and feedback through so that cancellation signals and progress reporting propagate correctly.

import processing
from qgis.core import QgsProcessingContext, QgsProcessingFeedback


def processAlgorithm(
    self,
    parameters: dict,
    context: QgsProcessingContext,
    feedback: QgsProcessingFeedback,
) -> dict:
    """Run native dissolve, then apply a custom filter as a second pass."""
    source = self.parameterAsSource(parameters, "INPUT", context)
    if source is None:
        raise QgsProcessingException(self.invalidSourceError(parameters, "INPUT"))

    # First pass: delegate to the native dissolve algorithm
    dissolved = processing.run(
        "native:dissolve",
        {
            "INPUT": parameters["INPUT"],
            "FIELD": [],
            "OUTPUT": "TEMPORARY_OUTPUT",
        },
        context=context,
        feedback=feedback,
        is_child_algorithm=True,   # essential: prevents double-registration of outputs
    )

    if feedback.isCanceled():
        return {}

    # Second pass: custom logic on the dissolved result
    dissolved_layer = dissolved["OUTPUT"]
    # ... continue processing dissolved_layer ...
    return {"OUTPUT": dissolved_layer}

The is_child_algorithm=True flag is mandatory when calling processing.run() from inside another algorithm. Without it, QGIS registers the intermediate output as a top-level project layer, which breaks modeler chaining and pollutes the layer panel.

Exposing Configurable Expressions #

For algorithms that need to evaluate field expressions — rather than just read raw attribute values — use QgsProcessingParameterExpression paired with QgsExpression evaluation inside processAlgorithm(). This pattern gives end users the full QGIS expression engine without requiring code changes.

from qgis.core import (
    QgsExpression,
    QgsExpressionContext,
    QgsExpressionContextUtils,
    QgsProcessingParameterExpression,
)


def initAlgorithm(self, config: dict | None = None) -> None:
    self.addParameter(
        QgsProcessingParameterExpression(
            "SCORE_EXPR",
            "Suitability score expression",
            parentLayerParameterName="INPUT",
            optional=True,
        )
    )


def _build_expression_context(
    self,
    source,
    context: QgsProcessingContext,
) -> QgsExpressionContext:
    """Construct an expression context scoped to the source layer."""
    expr_context = QgsExpressionContextUtils.globalProjectLayerScopes(
        context.project(), source
    )
    return QgsExpressionContext(expr_context)

Headless Batch Execution and CI/CD Integration #

The same algorithm that runs interactively in the GUI executes identically via the qgis_process CLI. This is the foundation for headless spatial ETL pipelines and CI/CD data validation workflows.

# Headless execution — no display required
qgis_process run enterprise_provider:custombufferfilter \
  --INPUT=/data/parcels.gpkg \
  --BUFFER_DIST=50 \
  --OUTPUT=/data/buffered_parcels.gpkg

# List all algorithms from a provider
qgis_process list | grep enterprise_provider

# Inspect parameter schema as JSON
qgis_process help enterprise_provider:custombufferfilter --json

For automated spatial validation pipelines, pair qgis_process with pytest-qgis to run deterministic geometry assertions against known fixtures. The asynchronous task execution guide covers how to wrap processing.run() calls in QgsTask subclasses when you need progress feedback in the QGIS task manager without blocking the GUI.

Pitfalls and Debugging #

None returned from parameterAsSource(): The source layer failed to load — commonly a broken file path, missing OGR driver, or a layer that was removed from the project between UI validation and execution. Always raise QgsProcessingException(self.invalidSourceError(...)) immediately.

Sink geometry type mismatch: Passing QgsWkbTypes.Polygon to a sink opened for Point input causes silent feature drops or a cryptic provider error. When your algorithm changes geometry type (e.g. point → buffer polygon), pass the correct output type explicitly to parameterAsSink() rather than forwarding source.wkbType().

is_child_algorithm omitted on nested processing.run(): The intermediate output is registered as a top-level project layer, which breaks modeler chaining. Always pass is_child_algorithm=True for any processing.run() call inside another algorithm’s body.

CRS mismatch between source and destination: Never assume input and output layers share the same coordinate reference system. If your algorithm accepts two input layers, transform one to the other’s CRS using QgsCoordinateTransform before any spatial operation that compares or combines geometries.

UI interaction from processAlgorithm(): Accessing iface, QgsMapCanvas, or any QWidget from the execution body causes Qt thread-safety violations that produce intermittent crashes. All UI interaction must go through QgsProcessingFeedback (for messages) or be deferred to a post-execution hook.

FastInsert with duplicate feature IDs: QgsFeatureSink.FastInsert skips the duplicate-ID check for performance. When cloning existing features into a new sink this is safe. When merging features from multiple sources, reset feature IDs with feature.setId(QgsFeature.noAttributes) beforehand.

Memory leak from context layer store: If you open a QgsVectorLayer inside processAlgorithm() using the file path constructor instead of a source parameter, the framework does not manage its lifecycle. Add it to context.temporaryLayerStore() or ensure you delete it before the method returns.

Algorithm not appearing in Toolbox after registration: Usually caused by a duplicate id() clash with an existing provider, or a Python exception inside loadAlgorithms() that the framework silently swallows. Enable QGIS debug output (QGIS_DEBUG=1) and check QgsMessageLog for provider load errors.

Conclusion #

The QgsProcessingAlgorithm contract — metadata methods, initAlgorithm(), and processAlgorithm() — is a minimal, well-defined interface that lets the same spatial logic run in four execution environments without modification. Correctness depends on three disciplines: using destination sinks (never hardcoded paths) for output management, honouring feedback.isCanceled() at every loop iteration, and keeping processAlgorithm() free of UI interaction. Composing algorithms via processing.run(..., is_child_algorithm=True) enables modular pipelines that integrate natively with the graphical modeler. For teams standardizing their toolset, the reusable PyQGIS processing algorithm template provides a production-ready starting point with consistent error handling, logging, and parameter validation patterns.

Related

• Plugin Development & UI Integration — parent guide covering the full plugin development lifecycle
• Plugin Lifecycle and Resource Management — provider registration, initGui() / unload() symmetry, and memory cleanup
• Asynchronous Task Execution with QgsTask — wrapping processing calls in background tasks to keep the GUI responsive
• Designing Qt Dialogs and Form Widgets — custom parameter widgets and validation dialogs for complex algorithm inputs
• Vector and Raster Data Access Patterns — optimized feature iteration, request objects, and raster block reading inside algorithm bodies