PyQGIS Engineering & Automation

Creating a Reusable PyQGIS Processing Algorithm Template

To build a production-grade PyQGIS processing algorithm, subclass QgsProcessingAlgorithm, implement the mandatory metadata and I/O methods, and register the…

To build a production-grade PyQGIS processing algorithm, subclass QgsProcessingAlgorithm, implement the mandatory metadata and I/O methods, and register the instance with the QGIS processing framework. The architecture must strictly separate execution logic from UI rendering, route progress and cancellation signals through QgsProcessingFeedback, and validate inputs before touching the feature iterator. Below is a complete, QGIS 3.x-compatible template that handles parameter definition, safe feature iteration, output sink creation, and graceful error recovery. You can drop this directly into a custom plugin provider or a standalone automation script.

Production-Ready Template Code

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

class ReusableTemplateAlgorithm(QgsProcessingAlgorithm):
    # --- Parameter Keys ---
    INPUT = 'INPUT'
    OUTPUT = 'OUTPUT'
    THRESHOLD = 'THRESHOLD'

    def name(self):
        return 'reusable_template'

    def displayName(self):
        return 'Reusable Processing Template'

    def group(self):
        return 'Custom Automation'

    def groupId(self):
        return 'custom_automation'

    def shortHelpString(self):
        return "A production-ready template for building custom PyQGIS processing algorithms. " \
               "See [Building Custom Processing Algorithms](/plugin-development-ui-integration/building-custom-processing-algorithms/) for advanced parameter chaining and model integration."

    def createInstance(self):
        return ReusableTemplateAlgorithm()

    # --- I/O Definition ---
    def initAlgorithm(self, config=None):
        self.addParameter(
            QgsProcessingParameterFeatureSource(
                self.INPUT,
                'Input Vector Layer',
                [QgsProcessing.TypeVectorAnyGeometry]
            )
        )
        self.addParameter(
            QgsProcessingParameterNumber(
                self.THRESHOLD,
                'Processing Threshold',
                type=QgsProcessingParameterNumber.Double,
                defaultValue=0.5,
                minValue=0.0,
                maxValue=1.0
            )
        )
        self.addParameter(
            QgsProcessingParameterVectorDestination(
                self.OUTPUT,
                'Output Vector Layer',
                type=QgsProcessing.TypeVectorAnyGeometry
            )
        )

    # --- Execution ---
    def processAlgorithm(self, parameters, context, feedback):
        # 1. Validate & resolve input source
        source = self.parameterAsSource(parameters, self.INPUT, context)
        if source is None:
            raise QgsProcessingException(self.invalidSourceError(parameters, self.INPUT))

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

        # 3. Prepare output sink (inherits schema, geometry type, and CRS from input)
        (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 with progress & cancellation handling
        total = 100.0 / source.featureCount() if source.featureCount() else 0
        features = source.getFeatures()
        processed_count = 0

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

            # Example business logic: filter by threshold or pass-through
            # Replace with your actual spatial/attribute operations
            if threshold > 0.0:
                # Simulate conditional processing
                pass

            new_feature = QgsFeature()
            new_feature.setGeometry(feature.geometry())
            new_feature.setAttributes(feature.attributes())
            sink.addFeature(new_feature, QgsFeatureSink.FastInsert)
            
            processed_count += 1
            feedback.setProgress(current * total)

        feedback.pushInfo(f"Successfully processed {processed_count} features.")
        return {self.OUTPUT: dest_id}

flowchart TD
    A["processAlgorithm()"] --> B["parameterAsSource() validate input"]
    B --> C["parameterAsSink() create output"]
    C --> D{"Next feature?"}
    D -->|"yes"| E{"feedback.isCanceled()?"}
    E -->|"yes"| Z["Break early"]
    E -->|"no"| F["sink.addFeature(); setProgress()"]
    F --> D
    D -->|"no"| G["return result dict"]
    Z --> G

Architecture Breakdown

Metadata & Instantiation

The name(), displayName(), group(), and groupId() methods define how the algorithm appears in the Processing Toolbox. QGIS uses these strings for localization, search indexing, and UI grouping. The createInstance() method is mandatory; the framework calls it to spawn fresh algorithm instances for each execution, preventing state leakage between runs.

Parameter Definition (I/O)

initAlgorithm() declares all inputs and outputs. Use QgsProcessingParameterFeatureSource for flexible layer acceptance and QgsProcessingParameterVectorDestination for memory-backed or file-backed outputs. The framework automatically handles file dialogs, layer dropdowns, and temporary storage routing. For deeper UI customization, consult the Plugin Development & UI Integration documentation to understand how processing providers bind to custom dialogs and batch interfaces.

Execution & Feedback Loop

processAlgorithm() is the core worker. It receives a parameters dictionary, a context object (containing project, transform, and temporary path settings), and a feedback object. Always:

  1. Resolve sources/sinks using parameterAsSource() and parameterAsSink().
  2. Check for None returns and raise QgsProcessingException with invalidSourceError() or invalidSinkError().
  3. Loop through source.getFeatures(), checking feedback.isCanceled() every iteration.
  4. Call feedback.setProgress() with a 0–100 float to update the progress bar.
  5. Return a dictionary mapping output keys to their resolved IDs.

This pattern aligns with the official QGIS PyQGIS Developer Cookbook, which emphasizes non-blocking execution and explicit sink management.

Registration & Integration

To make the algorithm discoverable, register it with the QGIS processing registry. In a plugin, override initGui() or your provider’s loadAlgorithms() method:

python
from qgis.core import QgsApplication, QgsProcessingProvider

class CustomProvider(QgsProcessingProvider):
    def loadAlgorithms(self):
        self.addAlgorithm(ReusableTemplateAlgorithm())

    def id(self):
        return "custom_automation_provider"

    def name(self):
        return "Custom Automation Provider"

# In your plugin's __init__.py or initGui():
provider = CustomProvider()
QgsApplication.processingRegistry().addProvider(provider)

For standalone scripts (outside QGIS Desktop), initialize the application context first:

python
from qgis.core import QgsApplication
QgsApplication.setPrefixPath("/usr", True)  # Adjust to your QGIS install path
qgs = QgsApplication([], False)
qgs.initQgis()
# Register provider, run algorithm via QgsProcessing.runAlgorithm()

Production Best Practices

  • Never mutate input layers in-place. Always write to a sink. The processing framework expects deterministic, side-effect-free execution.
  • Use QgsFeatureSink.FastInsert when adding features. It bypasses unnecessary geometry validation and accelerates bulk writes by 30–50%.
  • Handle CRS transformations explicitly. If your algorithm requires a specific projection, use QgsCoordinateTransform inside the loop or request reprojected inputs via QgsProcessingParameterFeatureSource flags.
  • Log with feedback.pushInfo() and feedback.pushWarning() instead of print(). The Processing UI captures these streams and surfaces them in the algorithm log panel.
  • Keep algorithms stateless. Store configuration in project variables or layer metadata, not class attributes. This prevents race conditions when running batch jobs or parallel models.
  • Test with QgsProcessingFeedback mocks. The QGIS API Reference documents all required method signatures. Mock the feedback object to verify progress reporting and cancellation behavior before UI deployment.

By following this structure, you ensure compatibility with QGIS 3.x modeler integration, batch processing, and headless automation pipelines.