Qgis Syntax Processing Scripts

qgis-syntax-processing-scripts

Quick Reference

Processing Framework Overview

Algorithm ID Format

Algorithm IDs follow the pattern provider:algorithm_name:

Key Parameter Types

---

Critical Warnings

NEVER call processing.run() without wrapping it in try/except QgsProcessingException. Algorithm failures raise exceptions that MUST be caught.

NEVER use hardcoded algorithm IDs from external providers (GRASS, SAGA, OTB) without first verifying availability via QgsApplication.processingRegistry().algorithmById(). These providers may not be installed.

NEVER show GUI elements (message boxes, dialogs) from within processAlgorithm(). Algorithms run in background threads by default. ALWAYS use the feedback object for all user communication.

NEVER manually load output layers inside processAlgorithm() using QgsProject.instance().addMapLayer(). ALWAYS return the output ID and let the Processing framework manage results.

NEVER use hardcoded temp paths like /tmp/result.gpkg. ALWAYS use 'memory:' or QgsProcessing.TEMPORARY_OUTPUT for intermediate results.

ALWAYS check feedback.isCanceled() at the top of every loop iteration in custom algorithms. Failure to check causes unresponsive cancellation.

ALWAYS report progress in custom algorithms using feedback.setProgress() with a percentage (0-100).

ALWAYS declare all outputs in initAlgorithm(). Undeclared outputs are invisible to the Processing framework and cannot be used in models or chains.

---

Decision Tree

Which approach to use?

Need to run an existing algorithm?
├── Yes → processing.run("provider:algorithm", params)
│   ├── Need result in project? → processing.runAndLoadResults()
│   └── Need non-blocking? → QgsProcessingAlgRunnerTask
│
Need to create a custom algorithm?
├── For a QGIS plugin? → QgsProcessingAlgorithm subclass + QgsProcessingProvider
├── Standalone script? → @alg decorator (saved to Processing Scripts folder)
└── Quick prototype? → @alg decorator

Need to run same algorithm on many inputs?
└── Batch processing loop with processing.run() per iteration

Output type selection

Where should the output go?
├── Intermediate result (not saved) → 'memory:' or QgsProcessing.TEMPORARY_OUTPUT
├── Persistent file → '/path/to/output.gpkg' (vectors) or '/path/to/output.tif' (rasters)
└── Add to project automatically → processing.runAndLoadResults()

---

Essential Patterns

Pattern 1: Running an Algorithm

import processing
from qgis.core import QgsProcessingException

try:
    result = processing.run("native:buffer", {
        'INPUT': layer,           # QgsVectorLayer, file path, or URI
        'DISTANCE': 100,
        'SEGMENTS': 5,
        'END_CAP_STYLE': 0,      # 0=Round, 1=Flat, 2=Square
        'JOIN_STYLE': 0,          # 0=Round, 1=Miter, 2=Bevel
        'MITER_LIMIT': 2,
        'DISSOLVE': False,
        'OUTPUT': 'memory:'
    })
    buffered_layer = result['OUTPUT']
except QgsProcessingException as e:
    print(f"Buffer failed: {e}")

Pattern 2: Running with Feedback

from qgis.core import QgsProcessingFeedback

feedback = QgsProcessingFeedback()
feedback.progressChanged.connect(lambda p: print(f"Progress: {p}%"))

result = processing.run("native:buffer", {
    'INPUT': layer,
    'DISTANCE': 100,
    'OUTPUT': 'memory:'
}, feedback=feedback)

Pattern 3: Background Execution

from qgis.core import (
    QgsApplication, QgsProcessingAlgRunnerTask,
    QgsProcessingContext, QgsProcessingFeedback, QgsProject
)

context = QgsProcessingContext()
context.setProject(QgsProject.instance())
feedback = QgsProcessingFeedback()

alg = QgsApplication.processingRegistry().algorithmById('native:buffer')
params = {'INPUT': layer, 'DISTANCE': 100, 'OUTPUT': 'memory:'}

task = QgsProcessingAlgRunnerTask(alg, params, context, feedback)

def on_complete(successful, results):
    if successful:
        output_layer = context.getMapLayer(results['OUTPUT'])
        QgsProject.instance().addMapLayer(output_layer)

task.executed.connect(on_complete)
QgsApplication.taskManager().addTask(task)

Pattern 4: Algorithm Chaining

import processing

# Step 1: Reproject to a projected CRS
reprojected = processing.run("native:reprojectlayer", {
    'INPUT': input_layer,
    'TARGET_CRS': 'EPSG:28992',
    'OUTPUT': 'memory:'
})

# Step 2: Buffer (pass previous output directly)
buffered = processing.run("native:buffer", {
    'INPUT': reprojected['OUTPUT'],
    'DISTANCE': 100,
    'OUTPUT': 'memory:'
})

# Step 3: Dissolve into final output
dissolved = processing.run("native:dissolve", {
    'INPUT': buffered['OUTPUT'],
    'OUTPUT': '/output/final_result.gpkg'
})

Pattern 5: Safe Algorithm Execution

from qgis.core import QgsApplication, QgsProcessingException
import processing

def safe_run(algorithm_id, params, context=None, feedback=None):
    """Run an algorithm after verifying it exists."""
    registry = QgsApplication.processingRegistry()
    if registry.algorithmById(algorithm_id) is None:
        raise QgsProcessingException(
            f'Algorithm "{algorithm_id}" not found. '
            f'Check that the required provider is installed and enabled.'
        )
    return processing.run(algorithm_id, params,
                          context=context, feedback=feedback)

Pattern 6: Batch Processing

import os
import processing

input_dir = '/data/input/'
output_dir = '/data/output/'
input_files = [f for f in os.listdir(input_dir) if f.endswith('.gpkg')]

for input_file in input_files:
    input_path = os.path.join(input_dir, input_file)
    output_path = os.path.join(output_dir, f'buffered_{input_file}')

    try:
        processing.run("native:buffer", {
            'INPUT': input_path,
            'DISTANCE': 50,
            'OUTPUT': output_path
        })
    except QgsProcessingException as e:
        print(f"Failed for {input_file}: {e}")

---

Common Operations

Discover Available Algorithms

from qgis.core import QgsApplication

registry = QgsApplication.processingRegistry()

# List all providers
for provider in registry.providers():
    print(provider.id(), provider.name())

# List algorithms from a provider
for alg in registry.algorithms():
    if alg.provider().id() == 'native':
        print(alg.id(), alg.displayName())

# Look up a specific algorithm
alg = registry.algorithmById('native:buffer')
if alg:
    print(alg.shortHelpString())

Custom Algorithm Required Methods

Register Custom Provider in Plugin

# In plugin __init__.py or main module
from qgis.core import QgsApplication
from .provider import MyPluginProvider

class MyPlugin:
    def __init__(self, iface):
        self.provider = None

    def initGui(self):
        self.provider = MyPluginProvider()
        QgsApplication.processingRegistry().addProvider(self.provider)

    def unload(self):
        QgsApplication.processingRegistry().removeProvider(self.provider)

Plugin metadata.txt MUST include: hasProcessingProvider=yes

Threading Flags

If your algorithm uses GUI elements or non-thread-safe APIs, set FlagNoThreading:

def flags(self):
    return super().flags() | QgsProcessingAlgorithm.FlagNoThreading

---

Reference Links

• references/methods.md -- API signatures for QgsProcessingAlgorithm, processing.run, parameter types, and providers
• references/examples.md -- Complete custom algorithm, @alg decorator script, batch processing, and chaining examples
• references/anti-patterns.md -- Processing pitfalls with WRONG/CORRECT comparisons

Official Sources

• https://docs.qgis.org/latest/en/docs/pyqgis_developer_cookbook/processing.html
• https://docs.qgis.org/latest/en/docs/user_manual/processing/index.html
• https://docs.qgis.org/latest/en/docs/user_manual/processing/toolbox.html
• https://qgis.org/pyqgis/master/core/QgsProcessingAlgorithm.html

anti-patterns.md — qgis-syntax-processing-scripts

AP-01: Unverified Algorithm ID

NEVER call processing.run() with an algorithm from an external provider without checking availability first.

# WRONG — crashes if GRASS is not installed
result = processing.run("grass:v.buffer", params)

# CORRECT — verify algorithm exists before running
from qgis.core import QgsApplication

registry = QgsApplication.processingRegistry()
alg = registry.algorithmById('grass:v.buffer')
if alg is None:
    feedback.reportError('GRASS provider not available. Install GRASS GIS.')
    return {}
result = processing.run('grass:v.buffer', params)

Why: External providers (GRASS, SAGA, OTB) require separate installations. Native (native:) and GDAL (gdal:) algorithms are ALWAYS available.

---

AP-02: Missing Exception Handling

NEVER call processing.run() without a try/except block.

# WRONG — unhandled exception crashes the script
result = processing.run("native:buffer", params)
layer = result['OUTPUT']

# CORRECT — catch processing exceptions
from qgis.core import QgsProcessingException

try:
    result = processing.run("native:buffer", params)
    layer = result['OUTPUT']
except QgsProcessingException as e:
    feedback.reportError(f'Buffer operation failed: {e}')
    return {}

Why: Invalid parameters, missing inputs, invalid geometries, and disk space issues all cause QgsProcessingException.

---

AP-03: No Cancellation Check in Loops

NEVER iterate over features without checking for cancellation.

# WRONG — user cannot cancel, no progress reporting
def processAlgorithm(self, parameters, context, feedback):
    for feature in source.getFeatures():
        # process feature...
        pass

# CORRECT — cancellation check and progress reporting
def processAlgorithm(self, parameters, context, feedback):
    total = 100.0 / source.featureCount() if source.featureCount() else 0
    for current, feature in enumerate(source.getFeatures()):
        if feedback.isCanceled():
            break
        # process feature...
        feedback.setProgress(int(current * total))

Why: Without cancellation checks, the user has no way to stop a long-running algorithm. Without progress, the UI appears frozen.

---

AP-04: Hardcoded Temporary Paths

NEVER use hardcoded temporary file paths for intermediate results.

# WRONG — /tmp may not exist on Windows, path conflicts possible
result = processing.run("native:buffer", {
    'INPUT': layer,
    'DISTANCE': 10,
    'OUTPUT': '/tmp/buffer_result.gpkg'
})

# CORRECT — let QGIS manage temporary storage
result = processing.run("native:buffer", {
    'INPUT': layer,
    'DISTANCE': 10,
    'OUTPUT': 'memory:'
})

# ALSO CORRECT — using the constant
from qgis.core import QgsProcessing

result = processing.run("native:buffer", {
    'INPUT': layer,
    'DISTANCE': 10,
    'OUTPUT': QgsProcessing.TEMPORARY_OUTPUT
})

Why: Hardcoded paths break cross-platform compatibility and risk file conflicts. Memory layers are automatically cleaned up.

---

AP-05: GUI Elements in processAlgorithm

NEVER use message boxes, dialogs, or any GUI widgets inside processAlgorithm().

# WRONG — will crash in background thread
def processAlgorithm(self, parameters, context, feedback):
    from qgis.PyQt.QtWidgets import QMessageBox
    QMessageBox.warning(None, 'Warning', 'Something happened')

# CORRECT — use the feedback object for all communication
def processAlgorithm(self, parameters, context, feedback):
    feedback.pushWarning('Something happened')
    feedback.pushInfo('Processing step completed')
    feedback.reportError('Critical error occurred')

Why: processAlgorithm() runs in a background thread by default. GUI operations from non-main threads cause crashes.

---

AP-06: Manual Layer Loading in Algorithm

NEVER load output layers into the project from within processAlgorithm().

# WRONG — bypasses the Processing framework's result handling
def processAlgorithm(self, parameters, context, feedback):
    # ... processing ...
    result_layer = QgsVectorLayer(output_path, 'result', 'ogr')
    QgsProject.instance().addMapLayer(result_layer)

# CORRECT — return the output and let the framework handle loading
def processAlgorithm(self, parameters, context, feedback):
    # ... processing ...
    return {self.OUTPUT: dest_id}

Why: The Processing framework manages output loading, including adding to the project, storing in context, and passing to downstream algorithms in chains/models.

---

AP-07: Missing createInstance()

NEVER forget to implement createInstance() in a custom algorithm.

# WRONG — missing createInstance(), algorithm will fail
class MyAlgorithm(QgsProcessingAlgorithm):
    def name(self):
        return 'myalgorithm'
    # ... other methods but no createInstance()

# CORRECT — ALWAYS implement createInstance()
class MyAlgorithm(QgsProcessingAlgorithm):
    def name(self):
        return 'myalgorithm'

    def createInstance(self):
        return MyAlgorithm()

Why: The Processing framework calls createInstance() to create copies of the algorithm for execution. Without it, the algorithm cannot be run.

---

AP-08: CRS Mismatch in Multi-Layer Operations

NEVER combine layers with different CRSs without reprojecting first.

# WRONG — intersection with mismatched CRSs gives incorrect results
result = processing.run("native:intersection", {
    'INPUT': layer_epsg4326,
    'OVERLAY': layer_epsg28992,
    'OUTPUT': 'memory:'
})

# CORRECT — reproject to matching CRS first
reprojected = processing.run("native:reprojectlayer", {
    'INPUT': layer_epsg4326,
    'TARGET_CRS': layer_epsg28992.crs(),
    'OUTPUT': 'memory:'
})

result = processing.run("native:intersection", {
    'INPUT': reprojected['OUTPUT'],
    'OVERLAY': layer_epsg28992,
    'OUTPUT': 'memory:'
})

Why: Processing algorithms execute in the input layer's CRS. Mismatched CRSs cause silent geometry errors, wrong results, or exceptions.

---

AP-09: Undeclared Algorithm Outputs

NEVER return output values from processAlgorithm() without declaring them in initAlgorithm().

# WRONG — COUNT is returned but not declared as an output
def initAlgorithm(self, config=None):
    self.addParameter(QgsProcessingParameterFeatureSink('OUTPUT', 'Output'))

def processAlgorithm(self, parameters, context, feedback):
    return {'OUTPUT': dest_id, 'COUNT': feature_count}

# CORRECT — declare all outputs
def initAlgorithm(self, config=None):
    self.addParameter(QgsProcessingParameterFeatureSink('OUTPUT', 'Output'))
    self.addOutput(QgsProcessingOutputNumber('COUNT', 'Feature count'))

def processAlgorithm(self, parameters, context, feedback):
    return {'OUTPUT': dest_id, 'COUNT': feature_count}

Why: Undeclared outputs are invisible to the Processing framework and cannot be used in models, chains, or the GUI.

---

AP-10: Using @alg Decorator in Plugins

NEVER use the @alg decorator for algorithms that belong to a plugin.

# WRONG — @alg scripts go to the Scripts provider, not your plugin provider
from qgis.processing import alg

@alg(name='my_plugin_tool', label='My Plugin Tool', group='My Plugin')
def my_tool(instance, parameters, context, feedback, inputs):
    pass

# CORRECT — use QgsProcessingAlgorithm subclass for plugin algorithms
class MyPluginTool(QgsProcessingAlgorithm):
    # Full class implementation registered via QgsProcessingProvider
    pass

Why: @alg scripts are ALWAYS added to the Processing Scripts provider. They cannot be added to a custom provider. Plugin algorithms MUST use the full subclass approach.

---

AP-11: Ignoring Source Validation

NEVER skip validation of source and sink parameters in processAlgorithm().

# WRONG — source could be None
def processAlgorithm(self, parameters, context, feedback):
    source = self.parameterAsSource(parameters, 'INPUT', context)
    for feature in source.getFeatures():  # NoneType error if source is None
        pass

# CORRECT — validate source before use
def processAlgorithm(self, parameters, context, feedback):
    source = self.parameterAsSource(parameters, 'INPUT', context)
    if source is None:
        raise QgsProcessingException(
            self.invalidSourceError(parameters, 'INPUT')
        )

    (sink, dest_id) = self.parameterAsSink(
        parameters, 'OUTPUT', context,
        source.fields(), source.wkbType(), source.sourceCrs()
    )
    if sink is None:
        raise QgsProcessingException(
            self.invalidSinkError(parameters, 'OUTPUT')
        )

Why: parameterAsSource() and parameterAsSink() return None when the input is invalid. Using invalidSourceError() and invalidSinkError() provides clear error messages.

---

AP-12: Thread-Unsafe Operations Without Flag

NEVER access iface, GUI elements, or non-thread-safe libraries in processAlgorithm() without setting FlagNoThreading.

# WRONG — iface access from background thread causes crash
def processAlgorithm(self, parameters, context, feedback):
    canvas = iface.mapCanvas()  # Thread-unsafe!

# CORRECT — set FlagNoThreading if you must access GUI
def flags(self):
    return super().flags() | QgsProcessingAlgorithm.FlagNoThreading

def processAlgorithm(self, parameters, context, feedback):
    canvas = iface.mapCanvas()  # Safe with FlagNoThreading

Why: Algorithms run in background threads by default. FlagNoThreading forces execution on the main thread, making GUI access safe but blocking the UI.

examples.md — qgis-syntax-processing-scripts

Example 1: Running a Built-in Algorithm

import processing
from qgis.core import QgsProcessingException, QgsVectorLayer

# Load input layer
layer = QgsVectorLayer('/data/buildings.gpkg|layername=buildings', 'buildings', 'ogr')

try:
    result = processing.run("native:buffer", {
        'INPUT': layer,
        'DISTANCE': 50,
        'SEGMENTS': 5,
        'END_CAP_STYLE': 0,   # 0=Round, 1=Flat, 2=Square
        'JOIN_STYLE': 0,       # 0=Round, 1=Miter, 2=Bevel
        'MITER_LIMIT': 2,
        'DISSOLVE': False,
        'OUTPUT': 'memory:'
    })
    buffered_layer = result['OUTPUT']
    print(f"Buffered layer has {buffered_layer.featureCount()} features")
except QgsProcessingException as e:
    print(f"Buffer failed: {e}")

Example 2: Running and Loading Results into Project

import processing

# Automatically adds the result to the QGIS project
result = processing.runAndLoadResults("native:buffer", {
    'INPUT': 'path/to/input.gpkg',
    'DISTANCE': 100,
    'OUTPUT': 'memory:'
})

Example 3: Algorithm with Feedback and Progress

import processing
from qgis.core import QgsProcessingFeedback, QgsProcessingContext, QgsProject

context = QgsProcessingContext()
context.setProject(QgsProject.instance())
feedback = QgsProcessingFeedback()

# Monitor progress
feedback.progressChanged.connect(lambda p: print(f"Progress: {p:.0f}%"))

try:
    result = processing.run("native:dissolve", {
        'INPUT': layer,
        'FIELD': ['category'],
        'OUTPUT': '/output/dissolved.gpkg'
    }, context=context, feedback=feedback)
except QgsProcessingException as e:
    print(f"Dissolve failed: {e}")

Example 4: Background Task Execution

from qgis.core import (
    QgsApplication, QgsProcessingAlgRunnerTask,
    QgsProcessingContext, QgsProcessingFeedback, QgsProject
)

context = QgsProcessingContext()
context.setProject(QgsProject.instance())
feedback = QgsProcessingFeedback()

alg = QgsApplication.processingRegistry().algorithmById('native:buffer')
params = {
    'INPUT': layer,
    'DISTANCE': 100,
    'OUTPUT': 'memory:'
}

task = QgsProcessingAlgRunnerTask(alg, params, context, feedback)

def on_complete(successful, results):
    if successful:
        output_layer = context.getMapLayer(results['OUTPUT'])
        QgsProject.instance().addMapLayer(output_layer)
        print("Background buffer complete")
    else:
        print("Background buffer failed")

task.executed.connect(on_complete)
QgsApplication.taskManager().addTask(task)

Example 5: Algorithm Chaining

import processing

# Step 1: Reproject to projected CRS for accurate distance calculations
reprojected = processing.run("native:reprojectlayer", {
    'INPUT': input_layer,
    'TARGET_CRS': 'EPSG:28992',
    'OUTPUT': 'memory:'
})

# Step 2: Buffer using projected coordinates
buffered = processing.run("native:buffer", {
    'INPUT': reprojected['OUTPUT'],
    'DISTANCE': 500,
    'SEGMENTS': 8,
    'OUTPUT': 'memory:'
})

# Step 3: Dissolve all buffers into one polygon
dissolved = processing.run("native:dissolve", {
    'INPUT': buffered['OUTPUT'],
    'OUTPUT': 'memory:'
})

# Step 4: Save final result to file
final = processing.run("native:fixgeometries", {
    'INPUT': dissolved['OUTPUT'],
    'OUTPUT': '/output/service_area.gpkg'
})

Example 6: Batch Processing Multiple Files

import os
import processing
from qgis.core import QgsProcessingException

input_dir = '/data/input_layers/'
output_dir = '/data/buffered/'
os.makedirs(output_dir, exist_ok=True)

input_files = [f for f in os.listdir(input_dir) if f.endswith('.gpkg')]

results = []
for input_file in input_files:
    input_path = os.path.join(input_dir, input_file)
    output_path = os.path.join(output_dir, f'buffered_{input_file}')

    try:
        result = processing.run("native:buffer", {
            'INPUT': input_path,
            'DISTANCE': 50,
            'SEGMENTS': 5,
            'OUTPUT': output_path
        })
        results.append((input_file, 'success'))
    except QgsProcessingException as e:
        results.append((input_file, f'failed: {e}'))

# Report
for filename, status in results:
    print(f"{filename}: {status}")

Example 7: Complete Custom QgsProcessingAlgorithm

from qgis.PyQt.QtCore import QCoreApplication
from qgis.core import (
    QgsProcessing,
    QgsProcessingAlgorithm,
    QgsProcessingParameterFeatureSource,
    QgsProcessingParameterFeatureSink,
    QgsProcessingParameterNumber,
    QgsProcessingParameterField,
    QgsProcessingException,
    QgsFeatureSink,
    QgsWkbTypes,
)


class BufferByFieldAlgorithm(QgsProcessingAlgorithm):
    """Buffers features using a numeric field value as the distance."""

    INPUT = 'INPUT'
    DISTANCE_FIELD = 'DISTANCE_FIELD'
    SEGMENTS = 'SEGMENTS'
    OUTPUT = 'OUTPUT'

    def name(self):
        return 'bufferbyfieldvalue'

    def displayName(self):
        return self.tr('Buffer by field value')

    def group(self):
        return self.tr('Custom vector tools')

    def groupId(self):
        return 'customvectortools'

    def shortHelpString(self):
        return self.tr(
            'Creates buffers around features using a numeric field '
            'value as the buffer distance for each feature.'
        )

    def tags(self):
        return ['buffer', 'variable', 'field', 'distance']

    def initAlgorithm(self, config=None):
        self.addParameter(
            QgsProcessingParameterFeatureSource(
                self.INPUT,
                self.tr('Input layer'),
                [QgsProcessing.TypeVectorAnyGeometry]
            )
        )
        self.addParameter(
            QgsProcessingParameterField(
                self.DISTANCE_FIELD,
                self.tr('Distance field'),
                parentLayerParameterName=self.INPUT,
                type=QgsProcessingParameterField.Numeric
            )
        )
        self.addParameter(
            QgsProcessingParameterNumber(
                self.SEGMENTS,
                self.tr('Segments'),
                type=QgsProcessingParameterNumber.Integer,
                defaultValue=5,
                minValue=1
            )
        )
        self.addParameter(
            QgsProcessingParameterFeatureSink(
                self.OUTPUT,
                self.tr('Buffered')
            )
        )

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

        field_name = self.parameterAsString(
            parameters, self.DISTANCE_FIELD, context
        )
        segments = self.parameterAsInt(parameters, self.SEGMENTS, context)

        (sink, dest_id) = self.parameterAsSink(
            parameters, self.OUTPUT, context,
            source.fields(),
            QgsWkbTypes.Polygon,
            source.sourceCrs()
        )
        if sink is None:
            raise QgsProcessingException(
                self.invalidSinkError(parameters, self.OUTPUT)
            )

        total = 100.0 / source.featureCount() if source.featureCount() else 0

        for current, feature in enumerate(source.getFeatures()):
            if feedback.isCanceled():
                break

            distance = feature[field_name]
            if distance is None or distance == 0:
                feedback.pushInfo(
                    f'Skipping feature {feature.id()}: no valid distance'
                )
                continue

            buffered_geom = feature.geometry().buffer(distance, segments)
            feature.setGeometry(buffered_geom)
            sink.addFeature(feature, QgsFeatureSink.FastInsert)

            feedback.setProgress(int(current * total))

        return {self.OUTPUT: dest_id}

    def createInstance(self):
        return BufferByFieldAlgorithm()

    def tr(self, string):
        return QCoreApplication.translate('Processing', string)

Example 8: @alg Decorator Script Algorithm

from qgis.processing import alg

@alg(name='my_buffer_script',
     label='My Buffer Script',
     group='My Scripts',
     group_label='My Script Group')
@alg.input(type=alg.SOURCE, name='INPUT', label='Input layer')
@alg.input(type=alg.DISTANCE, name='DISTANCE', label='Buffer distance',
           default=10.0)
@alg.input(type=alg.SINK, name='OUTPUT', label='Output layer')
def my_buffer(instance, parameters, context, feedback, inputs):
    """Buffer features by a fixed distance."""
    source = instance.parameterAsSource(parameters, 'INPUT', context)
    distance = instance.parameterAsDouble(parameters, 'DISTANCE', context)

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

    total = 100.0 / source.featureCount() if source.featureCount() else 0

    for current, feature in enumerate(source.getFeatures()):
        if feedback.isCanceled():
            break
        feature.setGeometry(feature.geometry().buffer(distance, 5))
        sink.addFeature(feature)
        feedback.setProgress(int(current * total))

    return {'OUTPUT': dest_id}

Important: @alg scripts are ALWAYS added to the Processing Scripts provider. They CANNOT be used in plugins. For plugin algorithms, ALWAYS use the full QgsProcessingAlgorithm subclass.

Example 9: Custom Processing Provider

from qgis.core import QgsProcessingProvider
from .buffer_by_field import BufferByFieldAlgorithm


class MyPluginProvider(QgsProcessingProvider):

    def loadAlgorithms(self):
        self.addAlgorithm(BufferByFieldAlgorithm())

    def id(self):
        return 'myplugin'

    def name(self):
        return 'My Plugin Tools'

    def longName(self):
        return 'My Plugin GIS Tools v1.0'

Register in Plugin

from qgis.core import QgsApplication
from .provider import MyPluginProvider


class MyPlugin:
    def __init__(self, iface):
        self.provider = None

    def initGui(self):
        self.provider = MyPluginProvider()
        QgsApplication.processingRegistry().addProvider(self.provider)

    def unload(self):
        QgsApplication.processingRegistry().removeProvider(self.provider)

Example 10: Safe Algorithm Runner with Provider Check

from qgis.core import QgsApplication, QgsProcessingException
import processing


def safe_run(algorithm_id, params, context=None, feedback=None):
    """Run a processing algorithm after verifying it exists."""
    registry = QgsApplication.processingRegistry()
    alg = registry.algorithmById(algorithm_id)
    if alg is None:
        raise QgsProcessingException(
            f'Algorithm "{algorithm_id}" not found. '
            f'Verify that the provider is installed and enabled.'
        )
    return processing.run(algorithm_id, params,
                          context=context, feedback=feedback)


# Usage
try:
    result = safe_run("grass:v.buffer", {
        'input': layer,
        'distance': 100,
        'output': 'memory:'
    })
except QgsProcessingException as e:
    print(f"Algorithm error: {e}")

Example 11: Discovering Algorithm Parameters

from qgis.core import QgsApplication

registry = QgsApplication.processingRegistry()
alg = registry.algorithmById('native:buffer')

if alg:
    print(f"Algorithm: {alg.displayName()}")
    print(f"Help: {alg.shortHelpString()}")
    print("\nParameters:")
    for param in alg.parameterDefinitions():
        print(f"  {param.name()} ({param.type()}): {param.description()}")
        if hasattr(param, 'defaultValue'):
            print(f"    Default: {param.defaultValue()}")
    print("\nOutputs:")
    for output in alg.outputDefinitions():
        print(f"  {output.name()} ({output.type()}): {output.description()}")

methods.md — qgis-syntax-processing-scripts

processing.run()

processing.run(
    algorithm_id: str,          # "provider:algorithm" format
    parameters: dict,           # Parameter name → value mapping
    context: QgsProcessingContext = None,
    feedback: QgsProcessingFeedback = None,
    is_child_algorithm: bool = False
) -> dict

Returns a dictionary with output keys mapping to results (layer references, file paths, numeric values).

processing.runAndLoadResults()

processing.runAndLoadResults(
    algorithm_id: str,
    parameters: dict,
    context: QgsProcessingContext = None,
    feedback: QgsProcessingFeedback = None,
    is_child_algorithm: bool = False
) -> dict

Same as processing.run() but automatically adds output layers to the current QGIS project.

---

QgsProcessingAlgorithm — Required Methods

name() -> str

Returns the unique algorithm identifier. MUST be lowercase with no spaces. This becomes the suffix in the algorithm ID (provider_id:name).

displayName() -> str

Returns the user-visible name shown in the Processing Toolbox and dialogs.

initAlgorithm(config: dict = None) -> None

Defines all input and output parameters using self.addParameter() and self.addOutput().

processAlgorithm(parameters: dict, context: QgsProcessingContext, feedback: QgsProcessingFeedback) -> dict

Core execution logic. MUST return a dictionary with output keys matching declared output parameter names.

createInstance() -> QgsProcessingAlgorithm

MUST return a new instance of the algorithm class. Required for the Processing framework to create copies.

def createInstance(self):
    return MyAlgorithm()

QgsProcessingAlgorithm — Optional Methods

group() -> str

Category display name in the Processing Toolbox.

groupId() -> str

Category identifier string (lowercase, no spaces).

shortHelpString() -> str

Help text displayed in the algorithm dialog.

tags() -> list[str]

Search keywords for algorithm discovery.

flags() -> QgsProcessingAlgorithm.Flags

Algorithm execution flags. Use FlagNoThreading for non-thread-safe algorithms:

def flags(self):
    return super().flags() | QgsProcessingAlgorithm.FlagNoThreading

canExecute() -> tuple[bool, str]

Returns whether the algorithm can currently execute and an error message if not.

checkParameterValues(parameters: dict, context: QgsProcessingContext) -> tuple[bool, str]

Custom parameter validation beyond type checking.

prepareAlgorithm(parameters: dict, context: QgsProcessingContext, feedback: QgsProcessingFeedback) -> bool

Pre-execution setup that runs on the main thread (before processAlgorithm).

postProcessAlgorithm(context: QgsProcessingContext, feedback: QgsProcessingFeedback) -> dict

Post-execution cleanup that runs on the main thread (after processAlgorithm).

---

QgsProcessingProvider

Required Methods

Optional Methods

---

QgsProcessingContext

QgsProcessingFeedback

---

Input Parameter Types — Full Reference

Output Parameter Types

Parameter Read Methods (in processAlgorithm)

---

QgsProcessing Constants

QgsProcessingParameterNumber Types

QgsProcessingParameterField Types

---

Common Algorithm IDs

Vector General

Raster