Processing script to load vector layer from PostGIS - connection string error

Question

I'm trying to build a processing plugin to directly load layers from PostGIS database where the user can define some parameters to import the right layer. Here is the code I'm working on :

from qgis.PyQt.QtCore import QCoreApplication
from qgis.core import (QgsProcessing,
                       QgsProcessingAlgorithm,
                       QgsProcessingException,
                       QgsProcessingParameterString,
                       QgsProcessingOutputString,
                       QgsProcessingParameterNumber,
                       QgsVectorLayer,
                       QgsDataSourceUri,
                       QgsProject)
from qgis import processing
from qgis.utils import *


class LoadLayer(QgsProcessingAlgorithm):
    """
    The following algorithm load a vector layer 
    from "xxx" postgis database
    """

    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 LoadLayer()

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

    def displayName(self):
        """
        Returns the translated algorithm name.
        """
        return self.tr('Load layers from PostGIS')

    def group(self):
        """
        Returns the name of the group this algorithm belongs to.
        """
        return self.tr('Load single vector layer')

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

    def shortHelpString(self):
        """
        Returns a localised short help string for the algorithm.
        """
        return self.tr('Charge les couches depuis PostGIS')

    def initAlgorithm(self, config=None):
        """
        Here we define the inputs and outputs of the algorithm.
        """
        self.addParameter(
            QgsProcessingParameterString(
                'HOST',
                self.tr('Host'),
                defaultValue = '172.16.0.116'
            )
        )
        self.addParameter(
            QgsProcessingParameterNumber(
                'PORT',
                self.tr('Port'),
                defaultValue = 5432
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'USERNAME',
                self.tr('Username'),
                defaultValue = 'xyz'
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'PASSWORD',
                self.tr('Password'),
                defaultValue = '***'
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'DATABASE',
                self.tr('Database'),
                defaultValue = 'abc'
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'SCHEMA',
                self.tr('schema'),
                defaultValue = 'schema'
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'LAYERNAME',
                self.tr('Layer name'),
                defaultValue = 'Name of the layer'
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'CONFIG',
                self.tr('Configuration'),
                defaultValue = 'Postgres'
            )
        )
        self.addOutput(
            QgsProcessingOutputString(
                'CONNEXIONSTR',
                self.tr('Imported layer')
            )
        )

    def processAlgorithm(self, parameters, context, feedback):
        
        host = self.parameterAsString (parameters,'HOST', context)

        port = self.parameterAsString (parameters, 'PORT', context)
        
        username = self.parameterAsString (parameters, 'USERNAME', context)

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

        database = self.parameterAsString (parameters, 'DATABASE', context)
        
        schema = self.parameterAsString (parameters, 'SCHEMA', context)
        
        layername = self.parameterAsString (parameters, 'LAYERNAME', context)
        
        config = self.parameterAsString (parameters, 'CONFIG', context)
        
        
        uri = QgsDataSourceUri()
        uri.setConnection(host, port, database, username, password)
        uri.setDataSource(schema, layername, "geom")

        layer = QgsVectorLayer(uri.uri(), layername, config)
        
        if feedback.isCanceled():
            return {}
        
        # Return the results
            return {'CONNEXIONSTR': layer}

However the ouput gives me an error message : TypeError: invalid result from LoadLayer.processAlgorithm(), NoneType cannot be converted to a C/C++ QVariantMap in this context

I've made some researches and I probably need a connection string starting with :

return {'CONNEXIONSTR' : "postgresql://{user}:{password}@{host}:{port}/{database}"} 

Previously, ThomasG77 helped me to structure my code and proposed this code: https://gis.stackexchange.com/posts/396815/revisions

It's currently not working. I probably need to also mention schema, layer name and geometry to successfully connect, but as I'm not a Python crack, I don't really know how to do it.

Answer 1

ThomasG77 provided me a very handy solution for the last missing part. The solution has been found through the PostGISExecuteAndLoadSQL.py file

Here the functional code :

from qgis.PyQt.QtCore import QCoreApplication
from qgis.core import (QgsProcessing,
                       QgsProcessingAlgorithm,
                       QgsProcessingException,
                       QgsProcessingContext,
                       QgsProcessingParameterString,
                       QgsProcessingOutputVectorLayer,
                       QgsProcessingOutputString,
                       QgsProcessingParameterNumber,
                       QgsVectorLayer,
                       QgsDataSourceUri,
                       QgsProject)
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 layer')

    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.addParameter(
            QgsProcessingParameterString(
                'SCHEMA',
                self.tr('schema'),
                defaultValue = 'public'
            )
        )
        self.addParameter(
            QgsProcessingParameterString(
                'LAYERNAME',
                self.tr('Layer name'),
                defaultValue = 'Name of the layer'
            )
        )

        self.addOutput(
            QgsProcessingOutputVectorLayer(
                'LAYER',
                self.tr('Vector layer'),
                QgsProcessing.TypeVectorAnyGeometry
            )
        )

    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)

        schema = self.parameterAsString (parameters, 'SCHEMA', context)
        
        layername = self.parameterAsString (parameters, 'LAYERNAME', context)

        uri = QgsDataSourceUri()
        uri.setConnection(host, port, database, username, password)
        uri.setDataSource(schema, layername, "geom")

        layer = QgsVectorLayer(uri.uri(), layername, "postgres")

        if feedback.isCanceled():
            return {}
        
        if not layer.isValid():
            raise QgsProcessingException(self.tr("""This layer is invalid!
                Please check the PostGIS log for error messages."""))

        context.temporaryLayerStore().addMapLayer(layer)
        context.addLayerToLoadOnCompletion(
            layer.id(),
            QgsProcessingContext.LayerDetails('SQL layer',
                                              context.project(),
                                              'LAYER'))

        # Return the results
        return {'LAYER': layer.id()}

Credits to ThomasG77