db_manager.py

# -*- coding: utf-8 -*-
"""
/***************************************************************************
 GuidedOfflineEditingPlugin layer_list.py
                                 A QGIS plugin
 Extend the built-in Offline Editing Plugin providing automated processes
 Generated by Plugin Builder: http://g-sherman.github.io/Qgis-Plugin-Builder/
                              -------------------
        begin                : 2019-06-08
        git sha              : $Format:%H$
        copyright            : (C) 2019 by Yann Voté
        email                : ygversil@lilo.org
 ***************************************************************************/

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
"""

from collections import namedtuple
from collections.abc import Callable
from urllib.parse import (urlunparse as build_url,
                          urlencode as build_url_query)

from qgis.core import QgsApplication


PG_PROJECT_STORAGE_TYPE = 'postgresql'
GPKG_PROJECT_STORAGE_TYPE = 'geopackage'


UrlParts = namedtuple('UrlParts', ('scheme', 'netloc', 'path', 'parameters',
                                   'query', 'fragment'))


def build_pg_project_url(authcfg, sslmode, host, port, dbname, schema,
                         project=None):
    """Returns a ``postgresql://`` URL with given parameters that can be used
    to open or list QGIS projects saved in PostgreSQL."""
    query_params = (
        ('authcfg', authcfg),
        ('sslmode', sslmode),
        ('dbname', dbname),
        ('schema', schema),
    )
    if project:
        query_params += (
            ('project', project),
        )
    return build_url(
        UrlParts(scheme=PG_PROJECT_STORAGE_TYPE,
                 netloc='{host}:{port}'.format(host=host, port=port),
                 path='',
                 parameters='',
                 query=build_url_query(query_params),
                 fragment='')
    )


def build_gpkg_project_url(gpkg_path, project=None):
    """Returns a ``geopackage://`` URL with given parameters that can be used
    to open or list QGIS projects saved in GeoPackage."""
    if project:
        query_params = (
            ('projectName', project),
        )
    return build_url(
        UrlParts(scheme=GPKG_PROJECT_STORAGE_TYPE,
                 netloc='',
                 path=str(gpkg_path),
                 parameters='',
                 query=build_url_query(query_params),
                 fragment='')
    )


class PostgresProjectDownloader(Callable):
    """Wraps just enough to connect to a PostgreSQL database using QGIS3 new
    authentication system and download list of editable layers."""

    def __init__(self, host='localhost', port=5432, dbname='qgis',
                 schema='qgis', authcfg='authorg', sslmode='disable'):
        self.url = build_pg_project_url(authcfg=authcfg, sslmode=sslmode,
                                        host=host, port=port, dbname=dbname,
                                        schema=schema)

    def __call__(self):
        pg_pstore = (QgsApplication.projectStorageRegistry().
                     projectStorageFromType(PG_PROJECT_STORAGE_TYPE))
        yield from sorted(pg_pstore.listProjects(self.url))