Load vector layers from PostGIS with a processing script

Question

I'm trying to build a processing script to automatize the loading of specific vector layers from postgis db. I wrote the following code lines which worked fine in python console :

tablename = "SUIVI_LH"
user = "avteam"
mdp = "password"

from qgis.core import QgsVectorLayer, QgsDataSourceUri

uri = QgsDataSourceUri()
# set host name, port, database name, username and password
uri.setConnection("172.16.0.116", "5432", "foresale", user, mdp)
# set database schema, table name and geometry column
uri.setDataSource ("infra", tablename, "geom")
# define a name for your layer (tablename by default) and "postgres" as data provider
vlayer=QgsVectorLayer(uri.uri(), tablename, "postgres")
if not vlayer.isValid:
    print ('Couche non valide. Vérifiez la validité des paramètres de connexion du script')
else:
    QgsProject.instance().addMapLayer(vlayer)

However I can't find the right method to make it a processing script. I tried those code lines from here : https://stackoverflow.com/questions/62510556/pyqgis-adding-layer-from-postgis-database-and-work-whith-it-in-qgis-interface

from qgis.core import *
from qgis.core import QgsProject
from PyQt5.QtCore import QFileInfo
from qgis.core import QgsVectorLayer, QgsDataSourceUri
from qgis.utils import *

def run_script(iface):
    uri = QgsDataSourceUri()
    uri.setConnection("172.16.0.116", "5432", "foresale", 
    "avteam", "password")
    uri.setDataSource("infra", "LH", "geom")

    layer = QgsVectorLayer(uri.uri(), "LH", "postgres")
    if not layer.isValid():
        print("Layer %s did not load" %layer.name())  
    QgsProject.instance().addMapLayer(layer)

    run_script()

I also tried this solution for multiple vectors layers which is even better : Zoom on a PostGIS vector layer in a new project with PyQGIS?

from qgis.core import *
from qgis.core import QgsProject
from PyQt5.QtCore import QFileInfo
from qgis.core import QgsVectorLayer, QgsDataSourceUri
from qgis.utils import *

def loadPgLayer(instance, vl, path, name, data_filter=""):
    """load a postgis vector layer"""
        vl.setDataSource(path, name, "postgres")
        if vl.isValid() and vl.setSubsetString(data_filter):
            instance.addMapLayer(vl)

        canvas = iface.mapCanvas()
        instance = QgsProject.instance()

        uri = QgsDataSourceUri()
        uri_params = dict(aHost="172.16.0.116",
                  aPort="5432",
                  aDatabase="foresale",
                  aUsername="avteam",
                  aPassword="password")
        uri.setConnection(**uri_params)

        vl_LH = QgsVectorLayer()
        vl_SUIVI_LH = QgsVectorLayer()

        name = "LH"
        uri.setDataSource("infra", name, "geom")
        path = uri.uri(False)
        loadPgLayer(instance, vl_LH, path, name)

        name = "SUIVI_LH"
        uri.setDataSource("infra", name, "geom")
        path = uri.uri(False)
        loadPgLayer(instance, vl_SUIVI_LH, path, name)

        canvas.refresh()
        canvas.waitWhileRendering()

Both scripts run but do not produce outputs. Any ideas why ?

Log :

QGIS version: 3.16.6-Hannover
QGIS code revision: bfd36fddc9
Qt version: 5.11.2
GDAL version: 3.1.4
GEOS version: 3.8.1-CAPI-1.13.3
PROJ version: Rel. 6.3.2, May 1st, 2020
Processing algorithm…
Algorithm 'model' starting…
Input parameters:
{ }

Model processed OK. Executed 0 algorithms total in 0 s.
Execution completed in 0.02 seconds
Results:
{}

Loading resulting layers
Algorithm 'model' finished

Answer 1

You need to create your own class inheriting from QgsProcessingAlgorithm

See official doc "Writing new Processing algorithms as Python scripts"

You may prefer "@alg decorator" approach that can avoid you to go through the QgsProcessingAlgorithm way. Method is also mentioned in above link from official doc.

You may also take a look at "PyQGIS 101: Writing a Processing script recipe about processing algorithm" for a shorter example

Edit: An example below to take only non spatial input

from qgis.PyQt.QtCore import QCoreApplication
from qgis.core import (QgsProcessing,
                       QgsProcessingAlgorithm,
                       QgsProcessingException,
                       QgsProcessingParameterString,
                       QgsProcessingOutputString,
                       QgsProcessingParameterNumber)
from qgis import processing


class ExampleProcessingAlgorithm(QgsProcessingAlgorithm):
    """
    This is an example algorithm that takes string inputs,
    to provide to DB connexion
    """

    def tr(self, string):
        """
        Returns a translatable string with the self.tr() function.
        """
        return QCoreApplication.translate('Processing', string)

    def createInstance(self):
        # Must return a new copy of your algorithm.
        return ExampleProcessingAlgorithm()

    def name(self):
        """
        Returns the unique algorithm name.
        """
        return 'postgresconnexion'

    def displayName(self):
        """
        Returns the translated algorithm name.
        """
        return self.tr('Return PostgreSQL connexion string')

    def group(self):
        """
        Returns the name of the group this algorithm belongs to.
        """
        return self.tr('Example scripts')

    def groupId(self):
        """
        Returns the unique ID of the group this algorithm belongs
        to.
        """
        return 'examplescripts'

    def shortHelpString(self):
        """
        Returns a localised short help string for the algorithm.
        """
        return self.tr('Example algorithm short description')

    def initAlgorithm(self, config=None):
        """
        Here we define the inputs and outputs of the algorithm.
        """
        # 'INPUT' is the recommended name for the main input
        # parameter.
        self.addParameter(
            QgsProcessingParameterString(
                'HOST',
                self.tr('Host'),
                defaultValue = 'localhost'
            )
        )
        self.addParameter(
            QgsProcessingParameterNumber(
                'PORT',
                self.tr('Port'),
                defaultValue = 5432
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'USERNAME',
                self.tr('Username'),
            )
        )
        # 'OUTPUT' is the recommended name for the main output
        # parameter.
        self.addParameter(
            QgsProcessingParameterString(
                'PASSWORD',
                self.tr('Password')
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'DATABASE',
                self.tr('Database')
            )
        )

        self.addOutput(
            QgsProcessingOutputString(
                'CONNEXION_STRING',
                self.tr('PostgreSQL connexion string')
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        """
        Here is where the processing itself takes place.
        """
        # First, we get the count of features from the INPUT layer.
        # This layer is defined as a QgsProcessingParameterFeatureSource
        # parameter, so it is retrieved by calling
        # self.parameterAsSource.
        host = self.parameterAsString (parameters,
                                                     'HOST',
                                                     context)

        # Retrieve the buffer distance and raster cell size numeric
        # values. Since these are numeric values, they are retrieved
        # using self.parameterAsDouble.
        port = self.parameterAsString (parameters, 'PORT',
                                            context)
        username = self.parameterAsString (parameters, 'USERNAME',
                                                context)

        password = self.parameterAsString (parameters, 'PASSWORD',
                                                context)

        database = self.parameterAsString (parameters, 'DATABASE',
                                                context)

        if feedback.isCanceled():
            return {}
        
        # Return the results
        return {'CONNEXION_STRING': f"postgres://{username}:{password}@{host}{'' if port else f':{port}'}/{database}"}

Answer 2

Further researches and help from ThomasG77 provided the solution. Here is the answer :

https://gis.stackexchange.com/a/398355/176653