How to Write a Custom QGIS Processing Script Algorithm in Python #

There is a large gap between a script that works and a tool other people can use. A script has hard-coded paths, no validation, no progress bar, and no way to appear in a model. A Processing algorithm has all four, plus a generated dialog you did not write, a command-line interface you did not write, and a stable id you can call from anywhere. Turning the first into the second is about eighty lines of well-understood boilerplate, and it is the highest-leverage refactor in QGIS automation.

Problem statement #

Your one-off script has become something colleagues ask for, and the requests are all the same shape:

• "Can you make it work on my layer?" — the paths are baked into the file.
• "Can I run it without opening a code editor?" — there is no dialog, and there should not have to be a hand-written one.
• "Can we put it in the nightly job?" — it has no id, so nothing can call it but a human.
• "Can it go in the model I built?" — the modeler only accepts registered algorithms.
• "It stopped and I don't know why." — no progress, no cancel, no error reporting beyond a traceback.
• "Which version did we run last quarter?" — no declared parameters, so no record of what was passed.

The goal: the same logic, wrapped as a QgsProcessingAlgorithm, available from the toolbox, the modeler, processing.run(), and qgis_process, with validation and progress for free.

Quick answer #

Subclass QgsProcessingAlgorithm, declare parameters in initAlgorithm, do the work in processAlgorithm.

from qgis.core import (
    QgsProcessing, QgsProcessingAlgorithm, QgsProcessingException,
    QgsProcessingParameterFeatureSource, QgsProcessingParameterFeatureSink,
    QgsProcessingParameterNumber, QgsFeatureSink, QgsField, QgsFields,
)
from qgis.PyQt.QtCore import QCoreApplication, QVariant


class AddAreaColumn(QgsProcessingAlgorithm):
    INPUT = "INPUT"
    ROUND_TO = "ROUND_TO"
    OUTPUT = "OUTPUT"

    def initAlgorithm(self, config=None):
        self.addParameter(QgsProcessingParameterFeatureSource(
            self.INPUT, self.tr("Input polygons"), [QgsProcessing.TypeVectorPolygon]))
        self.addParameter(QgsProcessingParameterNumber(
            self.ROUND_TO, self.tr("Decimal places"),
            QgsProcessingParameterNumber.Integer, defaultValue=2, minValue=0, maxValue=6))
        self.addParameter(QgsProcessingParameterFeatureSink(
            self.OUTPUT, self.tr("Output layer")))

    def processAlgorithm(self, parameters, context, feedback):
        source = self.parameterAsSource(parameters, self.INPUT, context)
        if source is None:
            raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT))
        places = self.parameterAsInt(parameters, self.ROUND_TO, context)

        fields = QgsFields(source.fields())
        fields.append(QgsField("area_m2", QVariant.Double))

        sink, dest_id = self.parameterAsSink(
            parameters, self.OUTPUT, context, fields, source.wkbType(), source.sourceCrs())

        count = source.featureCount() or 1
        for i, feature in enumerate(source.getFeatures()):
            if feedback.isCanceled():
                break
            attrs = feature.attributes()
            attrs.append(round(feature.geometry().area(), places))
            feature.setAttributes(attrs)
            sink.addFeature(feature, QgsFeatureSink.FastInsert)
            feedback.setProgress(int(100 * i / count))

        return {self.OUTPUT: dest_id}

    def name(self):        return "addareacolumn"
    def displayName(self): return self.tr("Add area column")
    def group(self):       return self.tr("Parcels")
    def groupId(self):     return "parcels"
    def createInstance(self): return AddAreaColumn()
    def tr(self, string):  return QCoreApplication.translate("Processing", string)

Save it through Processing → Scripts → Create New Script from Template and it appears in the toolbox immediately, callable as script:addareacolumn.

Step-by-step solution #

Declare parameters instead of reading them #

initAlgorithm is where a script stops being a script. Every parameter you declare gets a widget in the dialog, a validation rule, a slot in the model builder, and a key in the dictionary processing.run() accepts — from one line.

The parameter types you will use most:

QgsProcessingParameterField is the one worth noticing: it takes parentLayerParameterName="INPUT", so the dialog populates the field list from the layer the user just chose. That is real UI behaviour you get for a keyword argument.

Mark anything genuinely optional with optional=True and give it a defaultValue, so the algorithm can be called with a minimal dictionary.

Read parameters with the typed accessors #

Never index parameters directly. The parameterAs* methods resolve values that may be layers, layer ids, paths, expressions, or defaults, and they use context to do it.

source   = self.parameterAsSource(parameters, self.INPUT, context)
distance = self.parameterAsDouble(parameters, self.DISTANCE, context)
places   = self.parameterAsInt(parameters, self.ROUND_TO, context)
field    = self.parameterAsString(parameters, self.FIELD, context)
choice   = self.parameterAsEnum(parameters, self.METHOD, context)
crs      = self.parameterAsCrs(parameters, self.TARGET_CRS, context)
flag     = self.parameterAsBool(parameters, self.DISSOLVE, context)

parameterAsSource returning None means the input could not be resolved — raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT)) so the message names the parameter rather than the traceback naming a line number.

Write through a sink, not a file #

parameterAsSink returns a tuple: a QgsFeatureSink and a destination id.

sink, dest_id = self.parameterAsSink(
    parameters, self.OUTPUT, context,
    fields,               # QgsFields for the output
    source.wkbType(),     # geometry type
    source.sourceCrs(),   # CRS
)
if sink is None:
    raise QgsProcessingException(self.invalidSinkError(parameters, self.OUTPUT))

The sink abstracts where the output goes. The same code writes a GeoPackage, a scratch layer, or a PostGIS table depending on what the caller asked for — which is precisely why an algorithm composes into a model and a plain script does not. Return {self.OUTPUT: dest_id} and the framework hands the caller the right thing.

Adding a field means building a new QgsFields rather than mutating the source's:

fields = QgsFields(source.fields())
fields.append(QgsField("area_m2", QVariant.Double))

Use feedback for progress, messages, and cancellation #

total = 100.0 / source.featureCount() if source.featureCount() else 0
for current, feature in enumerate(source.getFeatures()):
    if feedback.isCanceled():
        break
    ...
    feedback.setProgress(int(current * total))

feedback.pushInfo(f"processed {current + 1} features")
feedback.pushWarning("3 features had null geometry and were skipped")

Checking isCanceled() inside the loop is what makes the dialog's Cancel button work — without it the button greys out and nothing happens, which users reasonably read as a hang. In a headless run the same object routes messages to your logger, as in running Processing algorithms from Python.

Give it a name, a group, and help #

def name(self):        return "addareacolumn"      # id: lowercase, no spaces, stable forever
def displayName(self): return self.tr("Add area column")
def group(self):       return self.tr("Parcels")
def groupId(self):     return "parcels"
def shortHelpString(self):
    return self.tr(
        "Adds an <b>area_m2</b> column computed from each polygon's geometry.\n\n"
        "The layer must be in a projected CRS — areas from a geographic CRS are "
        "in square degrees and meaningless."
    )

name() is the machine id and must never change once anything calls it. displayName() is the human label and can change freely. shortHelpString() becomes the panel beside the dialog and the text qgis_process help prints — the natural place to record the CRS caveat that would otherwise become a support question.

Register it so everything can find it #

For interactive use, save the file through Processing → Scripts → Add Script to Toolbox; it lands in the profile's processing/scripts folder and gets the id script:addareacolumn.

For automation, ship a provider with your code and register it explicitly — no profile involved:

from qgis.core import QgsApplication, QgsProcessingProvider

class ParcelTools(QgsProcessingProvider):
    def loadAlgorithms(self):
        self.addAlgorithm(AddAreaColumn())
    def id(self):        return "parceltools"
    def name(self):      return "Parcel tools"
    def longName(self):  return self.name()

provider = ParcelTools()                      # keep this reference alive!
QgsApplication.processingRegistry().addProvider(provider)

import processing
processing.run("parceltools:addareacolumn", {
    "INPUT": "data/parcels.gpkg|layername=parcels",
    "ROUND_TO": 1,
    "OUTPUT": "out/parcels_area.gpkg",
})

The comment is not decorative. If provider is garbage-collected, the registry holds a dangling pointer and the next lookup crashes the process rather than raising.

Code examples #

Example 1: The provider as a module #

"""parceltools/__init__.py — register the provider once, from anywhere."""
from qgis.core import QgsApplication, QgsProcessingProvider

from .add_area import AddAreaColumn
from .snap_slivers import SnapSlivers

_provider = None

class ParcelTools(QgsProcessingProvider):
    def loadAlgorithms(self):
        for alg in (AddAreaColumn(), SnapSlivers()):
            self.addAlgorithm(alg)
    def id(self):       return "parceltools"
    def name(self):     return "Parcel tools"
    def longName(self): return "Parcel tools"

def register():
    """Idempotent: safe to call from a script, a test, or a plugin."""
    global _provider
    registry = QgsApplication.processingRegistry()
    if registry.providerById("parceltools") is None:
        _provider = ParcelTools()
        registry.addProvider(_provider)
    return registry.providerById("parceltools")

Holding the provider in a module-level global is the simplest correct way to keep the reference alive for the life of the process.

Example 2: A field parameter driven by the input layer #

def initAlgorithm(self, config=None):
    self.addParameter(QgsProcessingParameterFeatureSource(
        self.INPUT, self.tr("Input layer"), [QgsProcessing.TypeVectorAnyGeometry]))
    self.addParameter(QgsProcessingParameterField(
        self.GROUP_FIELD, self.tr("Group by field"),
        parentLayerParameterName=self.INPUT,
        type=QgsProcessingParameterField.Any))
    self.addParameter(QgsProcessingParameterEnum(
        self.STAT, self.tr("Statistic"),
        options=["sum", "mean", "min", "max"], defaultValue=1))

Four lines of declaration produce a dialog where choosing a layer repopulates the field dropdown and the statistic is a combo box — behaviour that would be a hundred lines of Qt by hand.

Example 3: Validating beyond types #

Type validation is automatic; domain validation is yours. Override checkParameterValues so bad input is rejected before any work starts.

def checkParameterValues(self, parameters, context):
    source = self.parameterAsSource(parameters, self.INPUT, context)
    if source is not None and source.sourceCrs().isGeographic():
        return False, self.tr(
            "Input is in a geographic CRS; areas would be in square degrees. "
            "Reproject to a projected CRS first."
        )
    return super().checkParameterValues(parameters, context)

The message appears in the dialog before the user clicks Run, and in the exception text when called from Python — the encoding of a real domain rule in the one place everybody meets it. It is the same instinct as validating pipeline inputs automatically, applied at the algorithm boundary.

Example 4: An algorithm that wraps other algorithms #

Child algorithms inherit your feedback and context, so progress and cancellation flow through.

def processAlgorithm(self, parameters, context, feedback):
    import processing

    source = self.parameterAsSource(parameters, self.INPUT, context)
    distance = self.parameterAsDouble(parameters, self.DISTANCE, context)

    feedback.pushInfo("repairing geometries")
    fixed = processing.run("native:fixgeometries",
                           {"INPUT": parameters[self.INPUT], "OUTPUT": "TEMPORARY_OUTPUT"},
                           context=context, feedback=feedback, is_child_algorithm=True)["OUTPUT"]

    feedback.pushInfo("buffering")
    buffered = processing.run("native:buffer",
                              {"INPUT": fixed, "DISTANCE": distance, "SEGMENTS": 8,
                               "DISSOLVE": False, "OUTPUT": parameters[self.OUTPUT]},
                              context=context, feedback=feedback, is_child_algorithm=True)["OUTPUT"]

    return {self.OUTPUT: buffered}

is_child_algorithm=True is essential: it tells the framework this run is part of a larger one, so temporary outputs are not cleaned up before the parent finishes.

Example 5: Testing it without a GUI #

An algorithm is an ordinary class, so it tests like one.

import unittest
from pathlib import Path

from qgis_headless import qgis_headless
from parceltools import register
import processing
from qgis.core import QgsVectorLayer

class TestAddArea(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.ctx = qgis_headless()
        cls.app = cls.ctx.__enter__()
        register()

    @classmethod
    def tearDownClass(cls):
        cls.ctx.__exit__(None, None, None)

    def test_adds_area_column(self):
        out = processing.run("parceltools:addareacolumn", {
            "INPUT": "tests/fixtures/parcels_27700.gpkg",
            "ROUND_TO": 1,
            "OUTPUT": "TEMPORARY_OUTPUT",
        })["OUTPUT"]
        layer = out if isinstance(out, QgsVectorLayer) else QgsVectorLayer(out, "t", "ogr")
        self.assertIn("area_m2", [f.name() for f in layer.fields()])
        self.assertGreater(next(layer.getFeatures())["area_m2"], 0)

    def test_rejects_geographic_crs(self):
        with self.assertRaises(Exception):
            processing.run("parceltools:addareacolumn", {
                "INPUT": "tests/fixtures/parcels_4326.gpkg",
                "ROUND_TO": 1, "OUTPUT": "TEMPORARY_OUTPUT",
            })

The second test is the one that matters: it proves the validation rejects something. A check that has never refused anything has not been shown to work.

Explanation #

The design idea behind Processing is declarative parameters. Your algorithm says what it needs — a polygon source, an integer between 0 and 6, a sink — and the framework does everything that follows from that declaration: builds a dialog, validates types, exposes the algorithm in the model builder, generates command-line help, records the run in the history, and accepts a dictionary from Python. Every one of those is work you would otherwise write and maintain. The eighty lines of boilerplate are not overhead; they are the interface fee for a very large amount of free infrastructure.

The parameterAs* accessors are where that abstraction pays off in a way that is easy to miss. parameterAsSource might receive a QgsVectorLayer, a layer id from a model's previous step, a file path with a |layername= suffix, a URI with a selection restriction, or a default. Your code sees a QgsFeatureSource and iterates it. The same is true at the other end: parameterAsSink hides whether the caller wanted a GeoPackage, a scratch layer, or a PostGIS table. Writing to a path directly, as a plain script would, throws all of that away and is the single change that makes an algorithm uncomposable.

Sinks also explain a performance property worth knowing. A feature source streams — getFeatures() yields features without materialising the layer — and QgsFeatureSink.FastInsert skips the per-feature id bookkeeping you do not need when writing new features. An algorithm written this way handles a layer larger than memory without any special effort, which a read everything into a list script does not. That is the same reasoning behind chunked reads in the pure-Python world, covered in fixing memory errors with large files.

The last point is about where this sits relative to the rest of the cluster. A Processing model composes existing algorithms and cannot express logic. A script algorithm is logic, wrapped so a model can use it. The two are complementary, and the productive pattern in a mature setup is: custom algorithms for the parts with real decisions in them, a model to wire those together in an analyst-editable graph, and a Python batch loop around the model for files, errors, and reporting. Each layer does the thing it is good at, and each is testable on its own.

Edge cases or notes #

Keep a reference to your provider #

QgsApplication.processingRegistry().addProvider(provider) does not take ownership on the Python side. If the local variable goes out of scope the object is collected and the registry holds a dangling pointer — the symptom is a crash, not an exception. Store it in a module global or on a long-lived object.

name() is forever #

The id is what models, scripts, and scheduled jobs reference. Changing name() silently breaks every caller. Change displayName() freely; treat name() as a published API.

createInstance must return a new object #

The framework clones the algorithm for each run. Returning self produces state bleeding between concurrent runs and subtle wrong answers. It is always return MyAlgorithm().

Import from qgis.PyQt #

from qgis.PyQt.QtCore import QVariant, never from PyQt5.QtCore import QVariant. The shim resolves to whichever Qt binding the build uses, so the same file keeps working across a Qt5-to-Qt6 QGIS transition.

Script algorithms in the profile versus a provider in your repo #

The Scripts folder is right for a personal utility and wrong for anything deployed: it lives in a user profile a service account does not have. For automation, put the class in your repository and register a provider explicitly, exactly as in the examples above.

flags() for long-running or non-thread-safe work #

Override flags() to return super().flags() | QgsProcessingAlgorithm.FlagNoThreading if your algorithm must run on the main thread — anything touching a GUI object, or a library that is not thread-safe. Without it the desktop may run it in a background task and crash.

Progress with an unknown total #

source.featureCount() can return -1 for some providers. Guard the division, and fall back to feedback.setProgressText() messages instead of a percentage rather than reporting a nonsense number.

Internal links #

• How to Automate QGIS with Python (PyQGIS): The Complete Workflow
• How to Run QGIS Processing Algorithms from Python
• How to Build a QGIS Processing Model and Run It from Python
• How to Batch Process Layers in QGIS with PyQGIS
• How to Run a PyQGIS Script Headless Without Opening QGIS
• How to Validate Pipeline Inputs and Outputs Automatically
• How to Chain GIS Processing Steps into a Reusable Pipeline

FAQ #

What is the difference between a script algorithm and a plain PyQGIS script? #

A plain script runs top to bottom with whatever paths you typed. An algorithm declares its parameters, which gives it a generated dialog, type validation, progress and cancellation, a stable id, availability in the model builder and in qgis_process, and a dictionary-based Python API. The logic is the same; the difference is that one is callable by anything and the other only by you.

Where should I save my algorithm — the Scripts folder or my repository? #

The Scripts folder is fine for a personal tool you use interactively. For anything that runs on a schedule or that colleagues depend on, keep the class in your repository and register a QgsProcessingProvider explicitly, so the algorithm ships with the code and does not depend on a user profile.

Why does my algorithm crash QGIS after I register the provider? #

Almost certainly because the provider object was garbage-collected while the registry still referenced it. The registry does not keep the Python object alive. Assign it to a module-level variable, or to an attribute of something long-lived, and the crash disappears.

How do I add a field to the output layer? #

Copy the source's fields into a new QgsFields, append your QgsField, and pass that to parameterAsSink. Then for each feature take feature.attributes(), append the new value, and setAttributes before adding it to the sink. Mutating source.fields() directly does not do what it looks like.

How do I make the Cancel button work? #

Check feedback.isCanceled() inside your feature loop and break out when it returns True. The framework sets the flag; only your loop can act on it. Without the check the button appears to do nothing, which users report as a freeze.

Can my algorithm call other algorithms? #

Yes, and it should for anything already implemented. Call processing.run(...) with context=context, feedback=feedback, is_child_algorithm=True. That keeps progress and cancellation flowing through and stops the framework cleaning up temporary outputs before your parent algorithm has finished with them.

How do I test an algorithm without opening QGIS? #

Start a headless QgsApplication, register your provider, and call the algorithm with processing.run inside ordinary unittest or pytest tests, writing to TEMPORARY_OUTPUT and asserting on the result layer. Include at least one test that feeds it invalid input and asserts that it raises — a validation rule that has never rejected anything is untested.