qgis_server_light

qgis_server_light

debug

profile

profile_async(func)

A profiler which can be used as a decorator on ASYNC methods to produce profiles and call graphs.

The Callgraph might be inspected with QCacheGrind or KCacheGrind or sth. else.

Parameters:

• func –

  the wrapped method

Returns:

• –

  The wrapper

Source code in src/qgis_server_light/debug/profile.py

def profile_async(func):
    """A profiler which can be used as a decorator on ASYNC methods to produce
    profiles and call graphs.

    The Callgraph might be inspected with QCacheGrind or KCacheGrind or sth. else.

    Args:
        func: the wrapped method

    Returns:
        The wrapper
    """

    async def wrapper(*args, **kwargs):
        import cProfile
        import pstats

        import pyprof2calltree

        pr = cProfile.Profile()
        pr.enable()

        result = await func(*args, **kwargs)

        pr.disable()
        stats = pstats.Stats(pr)
        path = Path(".profile")
        path.mkdir(parents=True, exist_ok=True)
        stats_file = os.path.join(
            str(path), f"{func.__module__}.{func.__name__}.profile_data.prof"
        )
        callgrind_file = os.path.join(
            str(path), f"{func.__module__}.{func.__name__}.callgrind.out"
        )
        stats.dump_stats(stats_file)
        call_tree = pyprof2calltree.CalltreeConverter(stats)
        with open(callgrind_file, "w+") as f:
            call_tree.output(f)
        print(
            f"Profiling data saved to: {stats_file}, CallTree saved t: {callgrind_file}"
        )

        return result

    return wrapper

profile_sync(func)

A profiler which can be used as a decorator on SYNC methods to produce profiles and call graphs.

The Callgraph might be inspected with QCacheGrind or KCacheGrind or sth. else.

Parameters:

• func –

  the wrapped method

Returns:

• –

  The wrapper

Source code in src/qgis_server_light/debug/profile.py

def profile_sync(func):
    """A profiler which can be used as a decorator on SYNC methods to produce
    profiles and call graphs.

    The Callgraph might be inspected with QCacheGrind or KCacheGrind or sth. else.

    Args:
        func: the wrapped method

    Returns:
        The wrapper
    """

    def wrapper(*args, **kwargs):
        import cProfile
        import pstats

        import pyprof2calltree

        pr = cProfile.Profile()
        pr.enable()

        result = func(*args, **kwargs)

        pr.disable()
        stats = pstats.Stats(pr)
        path = Path(".profile")
        path.mkdir(parents=True, exist_ok=True)
        stats_file = os.path.join(
            str(path), f"{func.__module__}.{func.__name__}.profile_data.prof"
        )
        callgrind_file = os.path.join(
            str(path), f"{func.__module__}.{func.__name__}.callgrind.out"
        )
        stats.dump_stats(stats_file)
        call_tree = pyprof2calltree.CalltreeConverter(stats)
        with open(callgrind_file, "w+") as f:
            call_tree.output(f)
        print(
            f"Profiling data saved to: {stats_file}, CallTree saved t: {callgrind_file}"
        )

        return result

    return wrapper

exporter

api

allowed_extensions = ('qgz', 'qgs') module-attribute

allowed_output_formats = ('json', 'xml') module-attribute

app = Flask(__name__) module-attribute

data_path = os.environ.get('QSL_DATA_ROOT', None) module-attribute

exporter_host = os.environ.get('QSL_EXPORTER_API_HOST', '127.0.0.1') module-attribute

exporter_port = int(os.environ.get('QSL_EXPORTER_API_PORT', 5000)) module-attribute

qgs = QgsApplication([], False) module-attribute

api_export()

Source code in src/qgis_server_light/exporter/api.py

@app.route("/export", methods=["POST"])
def api_export():
    logging.getLogger().setLevel(logging.DEBUG)
    data_path = os.environ.get("QSL_DATA_ROOT")
    body = request.get_json()
    parser_config = ParserConfig(fail_on_unknown_properties=True)
    parameters: ExportParameters = DictDecoder(config=parser_config).decode(
        body, ExportParameters
    )
    serializer_config = SerializerConfig(indent="  ")

    # project file
    project_file = ""
    for extension in allowed_extensions:
        project_file = path.join(
            data_path, parameters.mandant, ".".join([parameters.project, extension])
        )
        print(f"testing project_file: {project_file}")
        if path.exists(project_file):
            print(f"project_file: {project_file} EXISTS")
            break
    if not path.exists(project_file):
        raise NotImplementedError(
            f"Project {parameters.project} from mandant {parameters.mandant} not found."
        )
    print(f"project_file: {project_file}")

    # output format
    if parameters.output_format.lower() not in allowed_output_formats:
        raise NotImplementedError(
            f"Allowed output formats are: {'|'.join(allowed_output_formats)} not => {parameters.output_format}"
        )
    output_format = parameters.output_format.lower()

    full_pg_service_config = Exporter.merge_dicts(
        create_full_pg_service_conf(), parameters.pg_service_configs_dict
    )

    # extract
    exporter = Exporter(
        qgis_project_path=project_file,
        unify_layer_names_by_group=bool(parameters.unify_layer_names_by_group),
        pg_service_configs=full_pg_service_config,
    )
    config = exporter.run()
    result = ExportResult(successful=False)

    content = None
    if output_format == "json":
        content = JsonSerializer(config=serializer_config).render(config)
    elif output_format == "xml":
        content = XmlSerializer(config=serializer_config).render(config)
    else:
        return Response(JsonSerializer().render(result), mimetype="text/json")
    if content:
        with open(
            path.join(
                data_path,
                parameters.mandant,
                ".".join([parameters.project, output_format]),
            ),
            mode="w+",
        ) as f:
            f.write(content)
    result.successful = True
    return Response(JsonSerializer().render(result), mimetype="text/json")

cli

allowed_extensions = ('qgz', 'qgs') module-attribute

allowed_output_formats = ('json', 'xml') module-attribute

qgs = QgsApplication([], False) module-attribute

cli() -> None

Just the central cli entry command. Currently, we don't use it, but its here for future content.

Source code in src/qgis_server_light/exporter/cli.py

@click.group
def cli() -> None:
    """
    Just the central cli entry command. Currently, we don't use it, but its here
    for future content.

    """
    pass

export(project: str, unify_layer_names_by_group: bool = False, output_format: str | None = None, pg_service_conf: str | None = None) -> None

Source code in src/qgis_server_light/exporter/cli.py

@click.option("--project", help="Absolute path to the QGIS project.")
@click.option(
    "--unify_layer_names_by_group",
    default=False,
    help="Use the full tree path to unify job_layer_definition names.",
)
@click.option(
    "--output_format",
    default="json",
    help=f"The desired output format. Allowed are {'|'.join(allowed_output_formats)}.",
)
@click.option(
    "--pg_service_conf",
    default=None,
    help="Absolute path to a pg_service.conf file to take connection information from.",
)
@cli.command(
    "export",
    context_settings={"max_content_width": 120},
    help=f"""
    Export a QGIS project ({"|".join(allowed_extensions)}) (1st argument) file to {"|".join(allowed_output_formats)} format.

    It takes into account the PGSERVICEFILE environment variable. The cli might be called with:

      PGSERVICEFILE=<absolute-path-to-pg_service.conf> python -m qgis_server_light.exporter.cli ...

    The pg_service.conf absolute path can be passed with parameter too. If this is done, the one out of
    environment will be joined with the passed one. The passed one overwrites values of the environment one.
    """,
)
def export(
    project: str,
    unify_layer_names_by_group: bool = False,
    output_format: str | None = None,
    pg_service_conf: str | None = None,
) -> None:
    logging.getLogger().setLevel(logging.DEBUG)
    serializer_config = SerializerConfig(indent="  ")
    if output_format is None:
        output_format = "json"
    if not project.lower().endswith(allowed_extensions):
        raise NotImplementedError(
            f"Allowed qgis project file extensions are: {'|'.join(allowed_extensions)} not => {project}"
        )
    if output_format.lower() not in allowed_output_formats:
        raise NotImplementedError(
            f"Allowed output formats are: {'|'.join(allowed_output_formats)} not => {output_format}"
        )
    full_pg_service_config = create_full_pg_service_conf(pg_service_conf)
    if os.path.isfile(project):
        exporter = Exporter(
            project,
            unify_layer_names_by_group=bool(unify_layer_names_by_group),
            pg_service_configs=full_pg_service_config,
        )
        config = exporter.run()
        if output_format == "json":
            click.echo(JsonSerializer(config=serializer_config).render(config))
        elif output_format == "xml":
            click.echo(XmlSerializer(config=serializer_config).render(config))

    else:
        raise AttributeError("Project file does not exist")

common

create_full_pg_service_conf(pg_service_conf: str | None = None) -> dict

Source code in src/qgis_server_light/exporter/common.py

def create_full_pg_service_conf(pg_service_conf: str | None = None) -> dict:
    full_pg_service_config_env = {}
    try:
        logging.info(
            "Loading pg_service configs from ENVIRONMENT variable PGSERVICEFILE:"
        )
        full_config = pgserviceparser.full_config()
        for section in full_config.sections():
            full_pg_service_config_env[section] = dict(full_config[section])
            logging.info(f"  loaded service: {section}")
    except Exception as e:
        logging.info(f"  No service config loaded")
        logging.debug(f"  {e}")
    full_pg_service_config_passed = {}
    if pg_service_conf:
        try:
            logging.info("Loading pg_service configs from passed path")
            full_config = pgserviceparser.full_config(
                conf_file_path=Path(pg_service_conf)
            )
            for section in full_config.sections():
                full_pg_service_config_passed[section] = dict(full_config[section])
                logging.info(f"  loaded service: {section}")
        except Exception as e:
            logging.info(f"  No service config loaded")
            logging.debug(f"  {e}")
    full_pg_service_config = Exporter.merge_dicts(
        full_pg_service_config_env, full_pg_service_config_passed
    )
    return full_pg_service_config

extract

Exporter

Source code in src/qgis_server_light/exporter/extract.py

class Exporter:
    def __init__(
        self,
        qgis_project_path: str,
        unify_layer_names_by_group=False,
        unify_layer_names_by_group_separator=".",
        pg_service_configs=None,
    ):
        self.unify_layer_names_by_group_separator = unify_layer_names_by_group_separator
        self.path = qgis_project_path
        self.unify_layer_names_by_group = unify_layer_names_by_group
        self.pg_service_configs = pg_service_configs or {}

        # prepare QGIS instances
        self.qgis_project = self.open_qgis_project(qgis_project_path)
        self.qgis_project_tree_root = self.qgis_project.layerTreeRoot()
        self.version, self.assembled_name = self.prepare_qgis_project_name(
            self.qgis_project
        )

        # prepare QSL interface instances
        self.qsl_tree = Tree()
        self.qsl_datasets = Datasets()
        self.qsl_project = Project(name=self.assembled_name, version=self.version)
        self.qsl_project_metadata = self.extract_metadata(self.qgis_project)
        self.qsl_config = Config(
            project=self.qsl_project,
            meta_data=self.qsl_project_metadata,
            tree=self.qsl_tree,
            datasets=self.qsl_datasets,
        )

    def run(self) -> Config:
        self.walk_qgis_project_tree(self.qgis_project_tree_root, [])
        return self.qsl_config

    def walk_qgis_project_tree(
        self,
        entity: QgsLayerTreeNode,
        path: list[str],
    ):
        """
        This is a highly recursive function which walks to the qgis job_layer_definition tree to extract all knowledge out
        of it. It is called from itself again for each level of group like elements which are found.

        Args:
            entity: The QGIS projects tree node which can be a QgsLayerTree, QgsLayerTreeGroup or
                QgsLayerTreeLayer.
            path: The path is a list of string which stores the information of the current tree path. This is
                used to construct a string for unifying job_layer_definition names by their tree path.
        """
        if isinstance(entity, QgsLayerTreeLayer):
            # this is a job_layer_definition, we can extract its information
            self.extract_save_layer(
                entity,
                path,
            )
        elif isinstance(entity, QgsLayerTreeGroup) or isinstance(entity, QgsLayerTree):
            # these are "group like" elements, we need to step into them one level deeper.
            short_name = self.get_group_short_name(entity)
            if short_name != "":
                # '' is the root of the tree, we don't want it to be part of the path
                path = path + [short_name]
            self.extract_group(
                entity,
                path,
            )
            for child in entity.children():
                # we handle all the children of the group like thing.
                self.walk_qgis_project_tree(
                    child,
                    path,
                )
        else:
            logging.error(
                f"This element was not expected while walking QGIS project tree: {entity}"
            )

    def extract_group(
        self,
        group: QgsLayerTreeGroup,
        path: list[str],
    ):
        """
        Collects data pertaining to a QGIS job_layer_definition tree group. Basically this translates a QgsLayerTreeGroup
        into a QGIS-Server-Light TreeGroup.

        Args:
            group: The group which is handled.
            path: The path is a list of string which stores the information of the current tree path. This is
                used to construct a string for unifying job_layer_definition names by their tree path.
        """
        children = []
        for child in group.children():
            if isinstance(child, QgsLayerTreeGroup):
                children.append(self.get_group_short_name(child))
            else:
                children.append(child.layer().id())
        self.qsl_tree.members.append(
            TreeGroup(
                id=self.get_group_short_name(group),
                name=self.get_group_short_name(group),
                children=children,
            )
        )
        self.qsl_datasets.group.append(
            Group(
                id=self.get_group_short_name(group),
                name=self.get_group_short_name(group),
                title=self.get_group_title(group),
            )
        )

    def extract_save_layer(
        self,
        child: QgsLayerTreeLayer,
        path: list[str],
        types_from_editor_widget: bool = False,
    ):
        """Save the given job_layer_definition to the output path."""
        layer = child.layer()
        layer_type = self.get_layer_type(layer)
        if layer_type is None:
            return
        decoded = self.decode_datasource(layer)
        short_name = self.short_name(self.get_layer_short_name(child), path)
        if layer.isSpatial():
            crs = self.create_qsl_crs_from_qgs_layer(layer)
            layer_extent = layer.extent()
            bbox_wgs84 = self.create_qsl_bbox_from_qgis_rectangle_wgs84(
                self.qgis_project, layer, layer_extent
            )
            bbox = self.create_qsl_bbox_from_qgis_rectangle_extent(layer_extent)
            is_spatial = True
        else:
            crs = None
            bbox_wgs84 = None
            bbox = None
            is_spatial = False
        if layer_type == "vector":
            if layer.providerType().lower() == "ogr":
                source = DataSource(ogr=self.create_qsl_source_ogr(decoded))
            elif layer.providerType().lower() == "postgres":
                source = DataSource(postgres=self.create_qsl_source_postgresql(decoded))
                source_path_parts = []
                DictEncoder().encode(source.postgres)
                for field in fields(source.postgres):
                    source_path_parts.append(
                        f"{field.name}={getattr(source.postgres, field.name)!r}"
                    )
            elif layer.providerType().lower() == "wfs":
                # TODO: Correctly implement source!
                source = WfsSource()
            else:
                logging.error(
                    f"Unknown provider type {layer.providerType().lower()} for job_layer_definition {layer.title() or layer.name()}"
                )
                return
            qsl_vector_dataset_fields = self.create_qsl_fields_from_qgis_field(layer)
            self.qsl_datasets.vector.append(
                Vector(
                    name=short_name,
                    title=layer.title() or layer.name(),
                    styles=self.create_style_list(layer),
                    driver=layer.providerType(),
                    bbox_wgs84=bbox_wgs84,
                    fields=qsl_vector_dataset_fields,
                    source=source,
                    id=layer.id(),
                    crs=crs,
                    bbox=bbox,
                    minimum_scale=layer.minimumScale(),
                    maximum_scale=layer.maximumScale(),
                    geometry_type_simple=layer.geometryType().name,
                    geometry_type_wkb=layer.wkbType().name,
                    is_spatial=is_spatial,
                )
            )
        elif layer_type == "raster":
            if layer.providerType() == "gdal":
                source = DataSource(gdal=self.create_qsl_source_gdal(decoded))
            elif layer.providerType() == "wms":
                if "tileMatrixSet" in decoded:
                    source = DataSource(wmts=self.create_qsl_source_wmts(decoded))
                else:
                    if decoded.get("type") == "xyz":
                        source = DataSource(xyz=self.create_qsl_source_xyz(decoded))
                    else:
                        source = DataSource(wms=self.create_qsl_source_wms(decoded))
            else:
                logging.error(f"Unknown provider type: {layer.providerType().lower()}")
                return
            if source is not None:
                self.qsl_datasets.raster.append(
                    Raster(
                        name=short_name,
                        title=layer.title() or layer.name(),
                        styles=self.create_style_list(layer),
                        driver=layer.providerType(),
                        bbox_wgs84=bbox_wgs84,
                        source=source,
                        id=layer.id(),
                        crs=crs,
                        bbox=bbox,
                        minimum_scale=layer.minimumScale(),
                        maximum_scale=layer.maximumScale(),
                        is_spatial=is_spatial,
                    )
                )
            else:
                logging.error(
                    f"Source was None this is not expected. Layer was: {short_name}, QGIS job_layer_definition ID:{layer.id()}"
                )
        elif layer_type == "custom":
            if layer.providerType().lower() in ["xyzvectortiles", "mbtilesvectortiles"]:
                source = DataSource(
                    vector_tile=self.create_qsl_source_vector_tiles(decoded)
                )
            else:
                logging.error(
                    f"Unknown provider type: {layer.providerType().lower()} Layer was: {short_name}, QGIS job_layer_definition ID:{layer.id()}"
                )
                # TODO: make this more configurable
                return
            self.qsl_datasets.custom.append(
                Custom(
                    name=short_name,
                    title=layer.title() or layer.name(),
                    styles=self.create_style_list(layer),
                    driver=layer.providerType(),
                    bbox_wgs84=bbox_wgs84,
                    source=source,
                    id=layer.id(),
                    crs=crs,
                    bbox=bbox,
                    minimum_scale=layer.minimumScale(),
                    maximum_scale=layer.maximumScale(),
                    is_spatial=is_spatial,
                )
            )
        else:
            logging.error(
                f'Unknown layer_type "{layer_type}" Layer was: {short_name}, QGIS job_layer_definition ID:{layer.id()}'
            )
            return

    def short_name(self, short_name: str, path: list[str]) -> str:
        """
        Decides if to use the short name itself or the unified version by the tree path.

        Args:
            short_name: The short name either of the group or the job_layer_definition.
            path: The path is a list of string which stores the information of the current tree path. This is
                used to construct a string for unifying job_layer_definition names by their tree path.

        Returns:
            The short name itself or its unified tree path.
        """
        if self.unify_layer_names_by_group:
            return self.create_unified_short_name(short_name, path)
        else:
            return short_name

    def create_unified_short_name(self, short_name: str, path: list[str]):
        """
        Creates the unified short name, separated by the configured separator.

        Args:
            short_name: The short name either of the group or the job_layer_definition.
            path: The path is a list of string which stores the information of the current tree path. This is
                used to construct a string for unifying job_layer_definition names by their tree path.

        Returns:

        """
        short_name_parts = path + [short_name]
        return self.unify_layer_names_by_group_separator.join(short_name_parts)

    def decode_datasource(self, layer: QgsMapLayer) -> dict:
        """
        Decodes a QGIS map job_layer_definition datasource into a dict and ensures that types are pythonic and pathes are
        clean for further usage.

        Args:
            layer: The job_layer_definition which the datasource should be extracted from.

        Returns:
            The decoded and cleaned datasource.
        """
        decoded = QgsProviderRegistry.instance().decodeUri(
            layer.providerType(), layer.dataProvider().dataSourceUri()
        )
        logging.debug(f"Layer source: {decoded}")
        for key in decoded:
            if str(decoded[key]) == "None":
                decoded[key] = None
            elif str(decoded[key]) == "NULL":
                decoded[key] = None
            else:
                decoded[key] = str(decoded[key])
            if key == "path":
                decoded[key] = decoded[key].replace(
                    f"{self.qgis_project.readPath('./')}/", ""
                )
        return decoded

    @staticmethod
    def create_qsl_field_from_qgis_field(
        field: QgsField, is_primary_key: bool
    ) -> Field:
        attribute_type_xml = Exporter.obtain_simple_types_from_qgis_field_xml(field)
        (
            editor_widget_type,
            editor_widget_type_wfs,
            editor_widget_type_json,
            editor_widget_type_json_format,
        ) = Exporter.obtain_editor_widget_type_from_qgis_field(field)
        (
            attribute_type_json,
            attribute_type_json_format,
        ) = Exporter.obtain_simple_types_from_qgis_field_json(field)
        return Field(
            is_primary_key=is_primary_key,
            name=field.name(),
            type=field.typeName(),
            type_wfs=attribute_type_xml,
            type_oapif=attribute_type_json,
            type_oapif_format=attribute_type_json_format,
            alias=field.alias() or field.name().title(),
            comment=field.comment(),
            nullable=is_primary_key and Exporter.obtain_nullable(field),
            length=Exporter.provide_field_length(field),
            precision=Exporter.provide_field_precision(field),
            editor_widget_type=editor_widget_type,
            editor_widget_type_wfs=editor_widget_type_wfs,
            editor_widget_type_oapif=editor_widget_type_json,
            editor_widget_type_oapif_format=editor_widget_type_json_format,
        )

    @staticmethod
    def obtain_nullable(field: QgsField):
        if not (
            field.constraints().constraints()
            == QgsFieldConstraints.Constraint.ConstraintNotNull
        ):
            return True
        return False

    @staticmethod
    def provide_field_length(field: QgsField) -> int | None:
        length = field.length()
        if length > 0:
            return length
        else:
            return None

    @staticmethod
    def provide_field_precision(field: QgsField) -> int | None:
        precision = field.precision()
        if precision > 0:
            return precision
        else:
            return None

    @staticmethod
    def create_qsl_fields_from_qgis_field(layer: QgsVectorLayer) -> List[Field]:
        fields = []
        pk_indexes = layer.dataProvider().pkAttributeIndexes()
        for field_index, field in enumerate(layer.fields().toList()):
            fields.append(
                Exporter.create_qsl_field_from_qgis_field(
                    field, (field_index in pk_indexes)
                )
            )
        return fields

    @staticmethod
    def obtain_simple_types_from_qgis_field_xml(field: QgsField) -> str:
        """

        Args:
            field: The field of an `QgsVectorLayer`.

        Returns:
            Unified type name regarding
            [XSD spec](https://www.w3.org/TR/xmlschema11-2/#built-in-primitive-datatypes)
            IMPORTANT: If type is not matched within the function it will be `string` always!
        """
        attribute_type = field.type()
        if attribute_type == QMetaType.Type.Int:
            return "int"
        elif attribute_type == QMetaType.Type.UInt:
            return "unsignedInt"
        elif attribute_type == QMetaType.Type.LongLong:
            return "long"
        elif attribute_type == QMetaType.Type.ULongLong:
            return "unsignedLong"
        elif attribute_type == QMetaType.Type.Double:
            if field.length() > 0 and field.precision() == 0:
                return "integer"
            else:
                return "decimal"
        elif attribute_type == QMetaType.Type.Bool:
            return "boolean"
        elif attribute_type == QMetaType.Type.QDate:
            return "date"
        elif attribute_type == QMetaType.Type.QTime:
            return "time"
        elif attribute_type == QMetaType.Type.QDateTime:
            return "dateTime"
        else:
            return "string"

    @staticmethod
    def get_group_title(group: QgsLayerTreeGroup) -> str:
        if group.customProperty("wmsTitle"):
            return group.customProperty("wmsTitle")
        elif hasattr(group, "groupLayer"):
            # since QGIS 3.38
            if group.groupLayer():
                if group.groupLayer().serverProperties():
                    if group.groupLayer().serverProperties().title():
                        return group.groupLayer().serverProperties().title()
        return group.name()

    @staticmethod
    def get_group_short_name(group: QgsLayerTreeGroup) -> str:
        if group.customProperty("wmsShortName"):
            return group.customProperty("wmsShortName")
        elif hasattr(group, "groupLayer"):
            # since QGIS 3.38
            if group.groupLayer():
                if group.groupLayer().serverProperties():
                    if group.groupLayer().serverProperties().shortName():
                        return group.groupLayer().serverProperties().shortName()
        short_name = Exporter.sanitize_name(group.name(), lower=True)
        if short_name == "_":
            # this is the tree root, we return empty string here
            return ""
        return short_name

    @staticmethod
    def sanitize_name(raw: str, lower: bool = False) -> str:
        """
        Transforms an arbitrary string into a WMS/WFS and URL compatible short name for a job_layer_definition or group.

        Steps:
        1. Unicode‑NFD → ASCII‑transliteration (removes umlauts/diacritics).
        2. All chars, which are NOT [A‑Za‑z0‑9_.‑], will be replaced by '_' ersetzen.
        3. Reduce multiple underscore '_' to a single one.
        4. Remove leading '_', '.', '-'.
        5. If the string is empty OR does not start with a letter OR not start with '_',
           a leading '_' will be added.
        6. Optional all will be converted to lowercase (lower=True).
        """
        # 1. cleaning to ASCII
        ascii_str = (
            unicodedata.normalize("NFKD", raw).encode("ascii", "ignore").decode()
        )
        # 2. not allowed → '_'
        ascii_str = re.sub(r"[^A-Za-z0-9_.-]+", "_", ascii_str)
        # 3. remove multiple '_'
        ascii_str = re.sub(r"_+", "_", ascii_str)
        # 4. remove trailing chars
        ascii_str = ascii_str.strip("._-")
        # 5. ensure first char is correct (mainly xml stuff and URL)
        if not ascii_str or not re.match(r"[A-Za-z_]", ascii_str[0]):
            ascii_str = "_" + ascii_str
        # 6. Optional lowercase
        if lower:
            ascii_str = ascii_str.lower()
        return ascii_str

    @staticmethod
    def obtain_editor_widget_type_from_qgis_field(
        field: QgsField,
    ) -> Tuple[str, str, str, str] | Tuple[str, None, None, None]:
        """
        We simply mimikri [QGIS Server here](https://github.com/qgis/QGIS/blob/de98779ebb117547364ec4cff433f062374e84a3/src/server/services/wfs/qgswfsdescribefeaturetype.cpp#L153-L192)

        TODO: This could be improved alot! Maybe we can also backport that to QGIS core some day?

        Args:
            field: The field of an `QgsVectorLayer`.

        Returns:
            Unified type name regarding
            [XSD spec](https://www.w3.org/TR/xmlschema11-2/#built-in-primitive-datatypes)
        """
        attribute_type = field.type()
        setup = field.editorWidgetSetup()
        config = setup.config()
        editor_widget_type = setup.type()
        if editor_widget_type == "DateTime":
            field_format = config.get(
                "field_format", QgsDateTimeFieldFormatter.defaultFormat(attribute_type)
            )
            if field_format == QgsDateTimeFieldFormatter.TIME_FORMAT:
                return editor_widget_type, "time", "string", "time"
            elif field_format == QgsDateTimeFieldFormatter.DATE_FORMAT:
                return editor_widget_type, "date", "string", "date"
            elif field_format == QgsDateTimeFieldFormatter.DATETIME_FORMAT:
                return editor_widget_type, "dateTime", "string", "date-time"
            elif field_format == QgsDateTimeFieldFormatter.QT_ISO_FORMAT:
                return editor_widget_type, "dateTime", "string", "date-time"
        elif editor_widget_type == "Range":
            if config.get("Precision"):
                config_precision = int(config["Precision"])
                if config_precision != field.precision():
                    if config_precision == 0:
                        return editor_widget_type, "integer", "integer", "int64"
                    else:
                        return editor_widget_type, "decimal", "number", "double"

        logging.error(
            f"Editor widget type was not handled as expected: {editor_widget_type}"
        )
        return editor_widget_type, None, None, None

    @staticmethod
    def obtain_simple_types_from_qgis_field_json(
        field: QgsField,
    ) -> Tuple[str, str] | Tuple[str, None]:
        """
        Delivers the type match for json based on the field QgsField type.

        Args:
            field: The field of an `QgsVectorLayer`.

        Returns:
            Unified type name and format in a tuple.
        """
        attribute_type = field.type()
        if attribute_type == QMetaType.Type.Int:
            return "integer", None
        elif attribute_type == QMetaType.Type.UInt:
            return "integer", "uint32"
        elif attribute_type == QMetaType.Type.LongLong:
            return "integer", "int64"
        elif attribute_type == QMetaType.Type.ULongLong:
            return "integer", "uint64"
        elif attribute_type == QMetaType.Type.Double:
            return "number", "double"
        elif attribute_type == QMetaType.Type.Float:
            return "number", "float"
        elif attribute_type == QMetaType.Type.Bool:
            return "boolean", None
        elif attribute_type == QMetaType.Type.QDate:
            return "string", "date"
        elif attribute_type == QMetaType.Type.QTime:
            return "string", "time"
        elif attribute_type == QMetaType.Type.QDateTime:
            return "string", "date-time"
        else:
            # we handle all unknown types as string. Since its for JavaScript, this should be safe.
            return "string", None

    @staticmethod
    def create_qsl_source_wms(datasource: dict) -> WmsSource:
        return WmsSource.from_qgis_decoded_uri(datasource)

    @staticmethod
    def create_qsl_source_vector_tiles(datasource: dict) -> VectorTileSource:
        return VectorTileSource.from_qgis_decoded_uri(datasource)

    @staticmethod
    def create_qsl_source_xyz(datasource: dict) -> XYZSource:
        return XYZSource.from_qgis_decoded_uri(datasource)

    @staticmethod
    def create_qsl_source_wmts(datasource: dict) -> WmtsSource:
        return WmtsSource.from_qgis_decoded_uri(datasource)

    @staticmethod
    def create_qsl_source_gdal(datasource: dict) -> GdalSource:
        return GdalSource.from_qgis_decoded_uri(datasource)

    @staticmethod
    def create_qsl_source_ogr(datasource: dict) -> OgrSource:
        return OgrSource.from_qgis_decoded_uri(datasource)

    def create_qsl_source_postgresql(self, datasource: dict) -> PostgresSource:
        config = datasource
        if datasource.get("service"):
            if self.pg_service_configs.get(datasource["service"]):
                service_config = self.pg_service_configs[datasource["service"]]
            else:
                service_config = {}
            if service_config == {}:
                logging.error(
                    f"""
                    There was a pg_service configuration in the project but it could not be found in
                    available configurations: {datasource["service"]}
                    Its highly possible that the exported project won't work.
                """
                )
            # merging pg_service content with config of qgis project (qgis project config overwrites
            # pg_service configs
            config = Exporter.merge_dicts(service_config, datasource)
        if config.get("username"):
            config["username"]
        elif config.get("user"):
            config["user"]
        else:
            raise LookupError(
                f"Configuration does not contain any info about the db user name {config}"
            )
        postgres_source = PostgresSource.from_qgis_decoded_uri(config)
        postgres_source.ssl_mode_text = self.decide_sslmode(postgres_source.sslmode)
        return postgres_source

    @staticmethod
    def merge_dicts(a, b):
        """
        Merges two dicts recursively, b values overwrites a values.

        Args:
            a: Dictionary which is the base.
            b: Dictionary which is merged in and whose values overwrites the one.

        Returns:
            The merged dict.
        """
        result = a.copy()
        for key, value in b.items():
            if (
                key in result
                and isinstance(result[key], dict)
                and isinstance(value, dict)
            ):
                result[key] = Exporter.merge_dicts(result[key], value)
            else:
                result[key] = value
        return result

    @staticmethod
    def decide_sslmode(ssl_mode: int) -> str:
        """
        Mapper to map ssl modes from QGIS to plain postgres.

        Args:
            ssl_mode: The ssl mode of the datasource.

        Returns:
            The string representation of the ssl mode as it is used by postgres connections.
        """
        return QgsDataSourceUri.encodeSslMode(int(ssl_mode))

    @staticmethod
    def create_qsl_crs_from_qgs_layer(layer: QgsMapLayer) -> Crs:
        """
        Translates the QGIS job_layer_definition crs information into an instance of the QGIS-Server-Light interface Crs
        instance.

        Args:
            layer: The job_layer_definition to take the crs information from.

        Returns:
            The instance of the QSL interface Crs.
        """
        layer_crs = layer.dataProvider().crs()
        return Crs(
            postgis_srid=layer_crs.postgisSrid(),
            auth_id=layer_crs.authid(),
            ogc_uri=layer_crs.toOgcUri(),
            ogc_urn=layer_crs.toOgcUrn(),
        )

    @staticmethod
    def get_layer_short_name(layer: QgsLayerTreeLayer) -> str:
        """
        This method determines which name is used as the short name of the job_layer_definition.

        Args:
            layer: The job_layer_definition which the short name should be derived from.

        Returns:
            The short name.
        """
        if layer.layer().shortName():
            return layer.layer().shortName()
        elif hasattr(layer.layer(), "serverProperties"):
            if layer.layer().serverProperties().shortName():
                return layer.layer().serverProperties().shortName()
        return layer.layer().id()

    @staticmethod
    def make_wgs84_geom_transform(project, layer) -> QgsCoordinateTransform:
        """
        Creates a QgisCoordinateTransform context to transform a job_layer_definition to EPSG:4326 aka wgs84.

        Args:
            project: The QGIS project instance.
            layer: The job_layer_definition which's extent should be reprojected.

        Returns:
            The QGIS transformation context.
        """
        source_crs = layer.crs()
        epsg_4326 = QgsCoordinateReferenceSystem("EPSG:4326")
        return QgsCoordinateTransform(source_crs, epsg_4326, project)

    @staticmethod
    def create_qsl_bbox_from_qgis_rectangle_wgs84(
        project: QgsProject, layer: QgsMapLayer, extent: QgsRectangle
    ) -> BBox:
        """
        Reprojects the job_layer_definition's extent using WGS84.

        Args:
            project: The QGIS project instance for projection context.
            layer: The job_layer_definition which for the projection context.
            extent: The extent which will be reprojected.

        Returns:
            The QSL bbox reprojected to WGS84.
        """
        transformation_context = Exporter.make_wgs84_geom_transform(project, layer)
        reprojected_extent = transformation_context.transform(extent)
        return Exporter.create_qsl_bbox_from_qgis_rectangle_extent(reprojected_extent)

    @staticmethod
    def create_qsl_bbox_from_qgis_rectangle_extent(extent: QgsRectangle) -> BBox:
        return BBox(
            x_min=extent.xMinimum(),
            x_max=extent.xMaximum(),
            y_min=extent.yMinimum(),
            y_max=extent.yMaximum(),
        )

    @staticmethod
    def get_layer_type(layer: QgsMapLayer) -> str | None:
        """
        Gets the type of the given Qgis job_layer_definition as a string if the type is supported. This is to flatten the
        understanding of layers from qgis into something we can handle.

        Args:
            layer: The job_layer_definition to decide the type for.

        Returns:
            "raster", "vector" or "custom" if a job_layer_definition matched and None otherwise.
        """
        if isinstance(layer, QgsRasterLayer):
            return "raster"
        elif isinstance(layer, QgsVectorLayer):
            return "vector"
        elif (
            isinstance(layer, QgsVectorTileLayer)
            or isinstance(layer, QgsTiledSceneLayer)
            or isinstance(layer, QgsPointCloudLayer)
            or isinstance(layer, QgsMeshLayer)
        ):
            return "custom"
        else:
            logging.error(f"Not implemented: {layer.type()}")
        return None

    @staticmethod
    def open_qgis_project(path_to_project: str) -> QgsProject:
        """


        Args:
            path_to_project: The absolute path on the file system where the project can be read from.

        Returns:
            The opened project (read).
        """
        project = QgsProject.instance()
        project.read(path_to_project)
        return project

    @staticmethod
    def prepare_qgis_project_name(project: QgsProject) -> tuple[str, str]:
        """


        Args:
            project: The instantiated QGIS project.

        Returns:
            Tuple of str version, name
        """
        # TODO: Find a good approach to recognize different "versions" of a project.
        name = project.baseName()
        parts = name.split(".")
        version = parts.pop(0)
        assembled_name = ".".join(parts)
        if assembled_name == "":
            assembled_name = project.title()
        return version, assembled_name

    @staticmethod
    def extract_metadata(project: QgsProject) -> MetaData:
        """
        Creates a QSL interface instance for metadate pulled out of the QGIS project.

        Args:
            project: The instantiated QGIS project.

        Returns:
            The metadata interface instance.
        """
        _meta = project.metadata()
        wms_entries = Exporter.get_project_server_entries(project, "wms")
        service = Service(**dict(sorted({**wms_entries}.items())))
        return MetaData(
            service=service,
            author=_meta.author(),
            categories=_meta.categories(),
            creationDateTime=_meta.creationDateTime().toPyDateTime().isoformat(),
            language=_meta.language(),
            links=[link.url for link in _meta.links()],
        )

    @staticmethod
    def get_project_server_entries(project, scope_or_scopes: Union[str, list]) -> dict:
        """
        Gets values from the fields displayed in QGIS under Project > Properties > Server.
        Returns a Dictionary holding all pairs of <key, value> found at the corresponding scopes.
        Example:
            given   scope_or_scope = "wms" (or: ["wms"])
            returns { <wms_key1>: <wms_key1_value>, <wms_key2>: <wms_key2_value> ... }
        For now the implementation supports only WMS fields but can be easily expanded by
        adding <key/values> to the Dictionary below.
        """
        supported_scopes = {
            "wms": {
                "scopes": [
                    ("WMSContactOrganization", "contact_organization"),
                    ("WMSContactMail", "contact_mail"),
                    ("WMSContactPerson", "contact_person"),
                    ("WMSContactPhone", "contact_phone"),
                    ("WMSContactPosition", "contact_position"),
                    ("WMSFees", "fees"),
                    ("WMSKeywordList", "keyword_list"),
                    ("WMSOnlineResource", "online_resource"),
                    ("WMSServiceAbstract", "service_abstract"),
                    ("WMSServiceTitle", "service_title"),
                    ("WMSUrl", "resource_url"),
                ],
                "keys": ["/"],
            }
        }

        scopes = (
            [scope_or_scopes] if isinstance(scope_or_scopes, str) else scope_or_scopes
        )

        for scope in scopes:
            if scope not in supported_scopes:
                supported = ", ".join(supported_scopes.keys())
                error_detail = f"This scope is not supported: {scope}. Supported scopes: {supported}"
                raise ValueError(error_detail)

            scope_entries = supported_scopes[scope]["scopes"]
            key_entries = supported_scopes[scope]["keys"]
            to_collect = zip_longest(
                scope_entries, key_entries, fillvalue=key_entries[0]
            )

            def collect(acc, pair):
                scope, key = pair
                qgis_scope_name, our_scope_name = scope

                if "list" in qgis_scope_name.lower():
                    # PyQGIS sometimes violates Liskov's substitution principle so naming tricks needed
                    list_as_text = ", ".join(
                        project.readListEntry(qgis_scope_name, key)[0]
                    )
                    acc.append((our_scope_name, list_as_text))
                else:
                    acc.append(
                        (our_scope_name, project.readEntry(qgis_scope_name, key)[0])
                    )

                return acc

            return dict(reduce(collect, to_collect, []))

    @staticmethod
    def create_style_list(qgs_layer: QgsMapLayer) -> List[Style]:
        style_names = qgs_layer.styleManager().styles()
        style_list = []
        for style_name in style_names:
            style_doc = QDomDocument()
            qgs_layer.styleManager().setCurrentStyle(style_name)
            qgs_layer.exportNamedStyle(style_doc)
            style_list.append(
                Style(
                    name=style_name,
                    definition=urlsafe_b64encode(
                        zlib.compress(style_doc.toByteArray())
                    ).decode(),
                )
            )
        return style_list

path = qgis_project_path instance-attribute

pg_service_configs = pg_service_configs or {} instance-attribute

qgis_project = self.open_qgis_project(qgis_project_path) instance-attribute

qgis_project_tree_root = self.qgis_project.layerTreeRoot() instance-attribute

qsl_config = Config(project=(self.qsl_project), meta_data=(self.qsl_project_metadata), tree=(self.qsl_tree), datasets=(self.qsl_datasets)) instance-attribute

qsl_datasets = Datasets() instance-attribute

qsl_project = Project(name=(self.assembled_name), version=(self.version)) instance-attribute

qsl_project_metadata = self.extract_metadata(self.qgis_project) instance-attribute

qsl_tree = Tree() instance-attribute

unify_layer_names_by_group = unify_layer_names_by_group instance-attribute

unify_layer_names_by_group_separator = unify_layer_names_by_group_separator instance-attribute

__init__(qgis_project_path: str, unify_layer_names_by_group=False, unify_layer_names_by_group_separator='.', pg_service_configs=None)

Source code in src/qgis_server_light/exporter/extract.py

def __init__(
    self,
    qgis_project_path: str,
    unify_layer_names_by_group=False,
    unify_layer_names_by_group_separator=".",
    pg_service_configs=None,
):
    self.unify_layer_names_by_group_separator = unify_layer_names_by_group_separator
    self.path = qgis_project_path
    self.unify_layer_names_by_group = unify_layer_names_by_group
    self.pg_service_configs = pg_service_configs or {}

    # prepare QGIS instances
    self.qgis_project = self.open_qgis_project(qgis_project_path)
    self.qgis_project_tree_root = self.qgis_project.layerTreeRoot()
    self.version, self.assembled_name = self.prepare_qgis_project_name(
        self.qgis_project
    )

    # prepare QSL interface instances
    self.qsl_tree = Tree()
    self.qsl_datasets = Datasets()
    self.qsl_project = Project(name=self.assembled_name, version=self.version)
    self.qsl_project_metadata = self.extract_metadata(self.qgis_project)
    self.qsl_config = Config(
        project=self.qsl_project,
        meta_data=self.qsl_project_metadata,
        tree=self.qsl_tree,
        datasets=self.qsl_datasets,
    )

create_qsl_bbox_from_qgis_rectangle_extent(extent: QgsRectangle) -> BBox staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_bbox_from_qgis_rectangle_extent(extent: QgsRectangle) -> BBox:
    return BBox(
        x_min=extent.xMinimum(),
        x_max=extent.xMaximum(),
        y_min=extent.yMinimum(),
        y_max=extent.yMaximum(),
    )

create_qsl_bbox_from_qgis_rectangle_wgs84(project: QgsProject, layer: QgsMapLayer, extent: QgsRectangle) -> BBox staticmethod

Reprojects the job_layer_definition's extent using WGS84.

Parameters:

• project (QgsProject) –

  The QGIS project instance for projection context.

• layer (QgsMapLayer) –

  The job_layer_definition which for the projection context.

• extent (QgsRectangle) –

  The extent which will be reprojected.

Returns:

• BBox –

  The QSL bbox reprojected to WGS84.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_bbox_from_qgis_rectangle_wgs84(
    project: QgsProject, layer: QgsMapLayer, extent: QgsRectangle
) -> BBox:
    """
    Reprojects the job_layer_definition's extent using WGS84.

    Args:
        project: The QGIS project instance for projection context.
        layer: The job_layer_definition which for the projection context.
        extent: The extent which will be reprojected.

    Returns:
        The QSL bbox reprojected to WGS84.
    """
    transformation_context = Exporter.make_wgs84_geom_transform(project, layer)
    reprojected_extent = transformation_context.transform(extent)
    return Exporter.create_qsl_bbox_from_qgis_rectangle_extent(reprojected_extent)

create_qsl_crs_from_qgs_layer(layer: QgsMapLayer) -> Crs staticmethod

Translates the QGIS job_layer_definition crs information into an instance of the QGIS-Server-Light interface Crs instance.

Parameters:

• layer (QgsMapLayer) –

  The job_layer_definition to take the crs information from.

Returns:

• Crs –

  The instance of the QSL interface Crs.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_crs_from_qgs_layer(layer: QgsMapLayer) -> Crs:
    """
    Translates the QGIS job_layer_definition crs information into an instance of the QGIS-Server-Light interface Crs
    instance.

    Args:
        layer: The job_layer_definition to take the crs information from.

    Returns:
        The instance of the QSL interface Crs.
    """
    layer_crs = layer.dataProvider().crs()
    return Crs(
        postgis_srid=layer_crs.postgisSrid(),
        auth_id=layer_crs.authid(),
        ogc_uri=layer_crs.toOgcUri(),
        ogc_urn=layer_crs.toOgcUrn(),
    )

create_qsl_field_from_qgis_field(field: QgsField, is_primary_key: bool) -> Field staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_field_from_qgis_field(
    field: QgsField, is_primary_key: bool
) -> Field:
    attribute_type_xml = Exporter.obtain_simple_types_from_qgis_field_xml(field)
    (
        editor_widget_type,
        editor_widget_type_wfs,
        editor_widget_type_json,
        editor_widget_type_json_format,
    ) = Exporter.obtain_editor_widget_type_from_qgis_field(field)
    (
        attribute_type_json,
        attribute_type_json_format,
    ) = Exporter.obtain_simple_types_from_qgis_field_json(field)
    return Field(
        is_primary_key=is_primary_key,
        name=field.name(),
        type=field.typeName(),
        type_wfs=attribute_type_xml,
        type_oapif=attribute_type_json,
        type_oapif_format=attribute_type_json_format,
        alias=field.alias() or field.name().title(),
        comment=field.comment(),
        nullable=is_primary_key and Exporter.obtain_nullable(field),
        length=Exporter.provide_field_length(field),
        precision=Exporter.provide_field_precision(field),
        editor_widget_type=editor_widget_type,
        editor_widget_type_wfs=editor_widget_type_wfs,
        editor_widget_type_oapif=editor_widget_type_json,
        editor_widget_type_oapif_format=editor_widget_type_json_format,
    )

create_qsl_fields_from_qgis_field(layer: QgsVectorLayer) -> List[Field] staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_fields_from_qgis_field(layer: QgsVectorLayer) -> List[Field]:
    fields = []
    pk_indexes = layer.dataProvider().pkAttributeIndexes()
    for field_index, field in enumerate(layer.fields().toList()):
        fields.append(
            Exporter.create_qsl_field_from_qgis_field(
                field, (field_index in pk_indexes)
            )
        )
    return fields

create_qsl_source_gdal(datasource: dict) -> GdalSource staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_source_gdal(datasource: dict) -> GdalSource:
    return GdalSource.from_qgis_decoded_uri(datasource)

create_qsl_source_ogr(datasource: dict) -> OgrSource staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_source_ogr(datasource: dict) -> OgrSource:
    return OgrSource.from_qgis_decoded_uri(datasource)

create_qsl_source_postgresql(datasource: dict) -> PostgresSource

Source code in src/qgis_server_light/exporter/extract.py

def create_qsl_source_postgresql(self, datasource: dict) -> PostgresSource:
    config = datasource
    if datasource.get("service"):
        if self.pg_service_configs.get(datasource["service"]):
            service_config = self.pg_service_configs[datasource["service"]]
        else:
            service_config = {}
        if service_config == {}:
            logging.error(
                f"""
                There was a pg_service configuration in the project but it could not be found in
                available configurations: {datasource["service"]}
                Its highly possible that the exported project won't work.
            """
            )
        # merging pg_service content with config of qgis project (qgis project config overwrites
        # pg_service configs
        config = Exporter.merge_dicts(service_config, datasource)
    if config.get("username"):
        config["username"]
    elif config.get("user"):
        config["user"]
    else:
        raise LookupError(
            f"Configuration does not contain any info about the db user name {config}"
        )
    postgres_source = PostgresSource.from_qgis_decoded_uri(config)
    postgres_source.ssl_mode_text = self.decide_sslmode(postgres_source.sslmode)
    return postgres_source

create_qsl_source_vector_tiles(datasource: dict) -> VectorTileSource staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_source_vector_tiles(datasource: dict) -> VectorTileSource:
    return VectorTileSource.from_qgis_decoded_uri(datasource)

create_qsl_source_wms(datasource: dict) -> WmsSource staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_source_wms(datasource: dict) -> WmsSource:
    return WmsSource.from_qgis_decoded_uri(datasource)

create_qsl_source_wmts(datasource: dict) -> WmtsSource staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_source_wmts(datasource: dict) -> WmtsSource:
    return WmtsSource.from_qgis_decoded_uri(datasource)

create_qsl_source_xyz(datasource: dict) -> XYZSource staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_qsl_source_xyz(datasource: dict) -> XYZSource:
    return XYZSource.from_qgis_decoded_uri(datasource)

create_style_list(qgs_layer: QgsMapLayer) -> List[Style] staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def create_style_list(qgs_layer: QgsMapLayer) -> List[Style]:
    style_names = qgs_layer.styleManager().styles()
    style_list = []
    for style_name in style_names:
        style_doc = QDomDocument()
        qgs_layer.styleManager().setCurrentStyle(style_name)
        qgs_layer.exportNamedStyle(style_doc)
        style_list.append(
            Style(
                name=style_name,
                definition=urlsafe_b64encode(
                    zlib.compress(style_doc.toByteArray())
                ).decode(),
            )
        )
    return style_list

create_unified_short_name(short_name: str, path: list[str])

Creates the unified short name, separated by the configured separator.

Parameters:

• short_name (str) –

  The short name either of the group or the job_layer_definition.

• path (list[str]) –

  The path is a list of string which stores the information of the current tree path. This is used to construct a string for unifying job_layer_definition names by their tree path.

Returns:

Source code in src/qgis_server_light/exporter/extract.py

def create_unified_short_name(self, short_name: str, path: list[str]):
    """
    Creates the unified short name, separated by the configured separator.

    Args:
        short_name: The short name either of the group or the job_layer_definition.
        path: The path is a list of string which stores the information of the current tree path. This is
            used to construct a string for unifying job_layer_definition names by their tree path.

    Returns:

    """
    short_name_parts = path + [short_name]
    return self.unify_layer_names_by_group_separator.join(short_name_parts)

decide_sslmode(ssl_mode: int) -> str staticmethod

Mapper to map ssl modes from QGIS to plain postgres.

Parameters:

• ssl_mode (int) –

  The ssl mode of the datasource.

Returns:

• str –

  The string representation of the ssl mode as it is used by postgres connections.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def decide_sslmode(ssl_mode: int) -> str:
    """
    Mapper to map ssl modes from QGIS to plain postgres.

    Args:
        ssl_mode: The ssl mode of the datasource.

    Returns:
        The string representation of the ssl mode as it is used by postgres connections.
    """
    return QgsDataSourceUri.encodeSslMode(int(ssl_mode))

decode_datasource(layer: QgsMapLayer) -> dict

Decodes a QGIS map job_layer_definition datasource into a dict and ensures that types are pythonic and pathes are clean for further usage.

Parameters:

• layer (QgsMapLayer) –

  The job_layer_definition which the datasource should be extracted from.

Returns:

• dict –

  The decoded and cleaned datasource.

Source code in src/qgis_server_light/exporter/extract.py

def decode_datasource(self, layer: QgsMapLayer) -> dict:
    """
    Decodes a QGIS map job_layer_definition datasource into a dict and ensures that types are pythonic and pathes are
    clean for further usage.

    Args:
        layer: The job_layer_definition which the datasource should be extracted from.

    Returns:
        The decoded and cleaned datasource.
    """
    decoded = QgsProviderRegistry.instance().decodeUri(
        layer.providerType(), layer.dataProvider().dataSourceUri()
    )
    logging.debug(f"Layer source: {decoded}")
    for key in decoded:
        if str(decoded[key]) == "None":
            decoded[key] = None
        elif str(decoded[key]) == "NULL":
            decoded[key] = None
        else:
            decoded[key] = str(decoded[key])
        if key == "path":
            decoded[key] = decoded[key].replace(
                f"{self.qgis_project.readPath('./')}/", ""
            )
    return decoded

extract_group(group: QgsLayerTreeGroup, path: list[str])

Collects data pertaining to a QGIS job_layer_definition tree group. Basically this translates a QgsLayerTreeGroup into a QGIS-Server-Light TreeGroup.

Parameters:

• group (QgsLayerTreeGroup) –

  The group which is handled.

• path (list[str]) –

  The path is a list of string which stores the information of the current tree path. This is used to construct a string for unifying job_layer_definition names by their tree path.

Source code in src/qgis_server_light/exporter/extract.py

def extract_group(
    self,
    group: QgsLayerTreeGroup,
    path: list[str],
):
    """
    Collects data pertaining to a QGIS job_layer_definition tree group. Basically this translates a QgsLayerTreeGroup
    into a QGIS-Server-Light TreeGroup.

    Args:
        group: The group which is handled.
        path: The path is a list of string which stores the information of the current tree path. This is
            used to construct a string for unifying job_layer_definition names by their tree path.
    """
    children = []
    for child in group.children():
        if isinstance(child, QgsLayerTreeGroup):
            children.append(self.get_group_short_name(child))
        else:
            children.append(child.layer().id())
    self.qsl_tree.members.append(
        TreeGroup(
            id=self.get_group_short_name(group),
            name=self.get_group_short_name(group),
            children=children,
        )
    )
    self.qsl_datasets.group.append(
        Group(
            id=self.get_group_short_name(group),
            name=self.get_group_short_name(group),
            title=self.get_group_title(group),
        )
    )

extract_metadata(project: QgsProject) -> MetaData staticmethod

Creates a QSL interface instance for metadate pulled out of the QGIS project.

Parameters:

• project (QgsProject) –

  The instantiated QGIS project.

Returns:

• MetaData –

  The metadata interface instance.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def extract_metadata(project: QgsProject) -> MetaData:
    """
    Creates a QSL interface instance for metadate pulled out of the QGIS project.

    Args:
        project: The instantiated QGIS project.

    Returns:
        The metadata interface instance.
    """
    _meta = project.metadata()
    wms_entries = Exporter.get_project_server_entries(project, "wms")
    service = Service(**dict(sorted({**wms_entries}.items())))
    return MetaData(
        service=service,
        author=_meta.author(),
        categories=_meta.categories(),
        creationDateTime=_meta.creationDateTime().toPyDateTime().isoformat(),
        language=_meta.language(),
        links=[link.url for link in _meta.links()],
    )

extract_save_layer(child: QgsLayerTreeLayer, path: list[str], types_from_editor_widget: bool = False)

Save the given job_layer_definition to the output path.

Source code in src/qgis_server_light/exporter/extract.py

def extract_save_layer(
    self,
    child: QgsLayerTreeLayer,
    path: list[str],
    types_from_editor_widget: bool = False,
):
    """Save the given job_layer_definition to the output path."""
    layer = child.layer()
    layer_type = self.get_layer_type(layer)
    if layer_type is None:
        return
    decoded = self.decode_datasource(layer)
    short_name = self.short_name(self.get_layer_short_name(child), path)
    if layer.isSpatial():
        crs = self.create_qsl_crs_from_qgs_layer(layer)
        layer_extent = layer.extent()
        bbox_wgs84 = self.create_qsl_bbox_from_qgis_rectangle_wgs84(
            self.qgis_project, layer, layer_extent
        )
        bbox = self.create_qsl_bbox_from_qgis_rectangle_extent(layer_extent)
        is_spatial = True
    else:
        crs = None
        bbox_wgs84 = None
        bbox = None
        is_spatial = False
    if layer_type == "vector":
        if layer.providerType().lower() == "ogr":
            source = DataSource(ogr=self.create_qsl_source_ogr(decoded))
        elif layer.providerType().lower() == "postgres":
            source = DataSource(postgres=self.create_qsl_source_postgresql(decoded))
            source_path_parts = []
            DictEncoder().encode(source.postgres)
            for field in fields(source.postgres):
                source_path_parts.append(
                    f"{field.name}={getattr(source.postgres, field.name)!r}"
                )
        elif layer.providerType().lower() == "wfs":
            # TODO: Correctly implement source!
            source = WfsSource()
        else:
            logging.error(
                f"Unknown provider type {layer.providerType().lower()} for job_layer_definition {layer.title() or layer.name()}"
            )
            return
        qsl_vector_dataset_fields = self.create_qsl_fields_from_qgis_field(layer)
        self.qsl_datasets.vector.append(
            Vector(
                name=short_name,
                title=layer.title() or layer.name(),
                styles=self.create_style_list(layer),
                driver=layer.providerType(),
                bbox_wgs84=bbox_wgs84,
                fields=qsl_vector_dataset_fields,
                source=source,
                id=layer.id(),
                crs=crs,
                bbox=bbox,
                minimum_scale=layer.minimumScale(),
                maximum_scale=layer.maximumScale(),
                geometry_type_simple=layer.geometryType().name,
                geometry_type_wkb=layer.wkbType().name,
                is_spatial=is_spatial,
            )
        )
    elif layer_type == "raster":
        if layer.providerType() == "gdal":
            source = DataSource(gdal=self.create_qsl_source_gdal(decoded))
        elif layer.providerType() == "wms":
            if "tileMatrixSet" in decoded:
                source = DataSource(wmts=self.create_qsl_source_wmts(decoded))
            else:
                if decoded.get("type") == "xyz":
                    source = DataSource(xyz=self.create_qsl_source_xyz(decoded))
                else:
                    source = DataSource(wms=self.create_qsl_source_wms(decoded))
        else:
            logging.error(f"Unknown provider type: {layer.providerType().lower()}")
            return
        if source is not None:
            self.qsl_datasets.raster.append(
                Raster(
                    name=short_name,
                    title=layer.title() or layer.name(),
                    styles=self.create_style_list(layer),
                    driver=layer.providerType(),
                    bbox_wgs84=bbox_wgs84,
                    source=source,
                    id=layer.id(),
                    crs=crs,
                    bbox=bbox,
                    minimum_scale=layer.minimumScale(),
                    maximum_scale=layer.maximumScale(),
                    is_spatial=is_spatial,
                )
            )
        else:
            logging.error(
                f"Source was None this is not expected. Layer was: {short_name}, QGIS job_layer_definition ID:{layer.id()}"
            )
    elif layer_type == "custom":
        if layer.providerType().lower() in ["xyzvectortiles", "mbtilesvectortiles"]:
            source = DataSource(
                vector_tile=self.create_qsl_source_vector_tiles(decoded)
            )
        else:
            logging.error(
                f"Unknown provider type: {layer.providerType().lower()} Layer was: {short_name}, QGIS job_layer_definition ID:{layer.id()}"
            )
            # TODO: make this more configurable
            return
        self.qsl_datasets.custom.append(
            Custom(
                name=short_name,
                title=layer.title() or layer.name(),
                styles=self.create_style_list(layer),
                driver=layer.providerType(),
                bbox_wgs84=bbox_wgs84,
                source=source,
                id=layer.id(),
                crs=crs,
                bbox=bbox,
                minimum_scale=layer.minimumScale(),
                maximum_scale=layer.maximumScale(),
                is_spatial=is_spatial,
            )
        )
    else:
        logging.error(
            f'Unknown layer_type "{layer_type}" Layer was: {short_name}, QGIS job_layer_definition ID:{layer.id()}'
        )
        return

get_group_short_name(group: QgsLayerTreeGroup) -> str staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def get_group_short_name(group: QgsLayerTreeGroup) -> str:
    if group.customProperty("wmsShortName"):
        return group.customProperty("wmsShortName")
    elif hasattr(group, "groupLayer"):
        # since QGIS 3.38
        if group.groupLayer():
            if group.groupLayer().serverProperties():
                if group.groupLayer().serverProperties().shortName():
                    return group.groupLayer().serverProperties().shortName()
    short_name = Exporter.sanitize_name(group.name(), lower=True)
    if short_name == "_":
        # this is the tree root, we return empty string here
        return ""
    return short_name

get_group_title(group: QgsLayerTreeGroup) -> str staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def get_group_title(group: QgsLayerTreeGroup) -> str:
    if group.customProperty("wmsTitle"):
        return group.customProperty("wmsTitle")
    elif hasattr(group, "groupLayer"):
        # since QGIS 3.38
        if group.groupLayer():
            if group.groupLayer().serverProperties():
                if group.groupLayer().serverProperties().title():
                    return group.groupLayer().serverProperties().title()
    return group.name()

get_layer_short_name(layer: QgsLayerTreeLayer) -> str staticmethod

This method determines which name is used as the short name of the job_layer_definition.

Parameters:

• layer (QgsLayerTreeLayer) –

  The job_layer_definition which the short name should be derived from.

Returns:

• str –

  The short name.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def get_layer_short_name(layer: QgsLayerTreeLayer) -> str:
    """
    This method determines which name is used as the short name of the job_layer_definition.

    Args:
        layer: The job_layer_definition which the short name should be derived from.

    Returns:
        The short name.
    """
    if layer.layer().shortName():
        return layer.layer().shortName()
    elif hasattr(layer.layer(), "serverProperties"):
        if layer.layer().serverProperties().shortName():
            return layer.layer().serverProperties().shortName()
    return layer.layer().id()

get_layer_type(layer: QgsMapLayer) -> str | None staticmethod

Gets the type of the given Qgis job_layer_definition as a string if the type is supported. This is to flatten the understanding of layers from qgis into something we can handle.

Parameters:

• layer (QgsMapLayer) –

  The job_layer_definition to decide the type for.

Returns:

• str | None –

  "raster", "vector" or "custom" if a job_layer_definition matched and None otherwise.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def get_layer_type(layer: QgsMapLayer) -> str | None:
    """
    Gets the type of the given Qgis job_layer_definition as a string if the type is supported. This is to flatten the
    understanding of layers from qgis into something we can handle.

    Args:
        layer: The job_layer_definition to decide the type for.

    Returns:
        "raster", "vector" or "custom" if a job_layer_definition matched and None otherwise.
    """
    if isinstance(layer, QgsRasterLayer):
        return "raster"
    elif isinstance(layer, QgsVectorLayer):
        return "vector"
    elif (
        isinstance(layer, QgsVectorTileLayer)
        or isinstance(layer, QgsTiledSceneLayer)
        or isinstance(layer, QgsPointCloudLayer)
        or isinstance(layer, QgsMeshLayer)
    ):
        return "custom"
    else:
        logging.error(f"Not implemented: {layer.type()}")
    return None

get_project_server_entries(project, scope_or_scopes: Union[str, list]) -> dict staticmethod

Gets values from the fields displayed in QGIS under Project > Properties > Server. Returns a Dictionary holding all pairs of found at the corresponding scopes. Example: given scope_or_scope = "wms" (or: ["wms"]) returns { : , : ... } For now the implementation supports only WMS fields but can be easily expanded by adding to the Dictionary below.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def get_project_server_entries(project, scope_or_scopes: Union[str, list]) -> dict:
    """
    Gets values from the fields displayed in QGIS under Project > Properties > Server.
    Returns a Dictionary holding all pairs of <key, value> found at the corresponding scopes.
    Example:
        given   scope_or_scope = "wms" (or: ["wms"])
        returns { <wms_key1>: <wms_key1_value>, <wms_key2>: <wms_key2_value> ... }
    For now the implementation supports only WMS fields but can be easily expanded by
    adding <key/values> to the Dictionary below.
    """
    supported_scopes = {
        "wms": {
            "scopes": [
                ("WMSContactOrganization", "contact_organization"),
                ("WMSContactMail", "contact_mail"),
                ("WMSContactPerson", "contact_person"),
                ("WMSContactPhone", "contact_phone"),
                ("WMSContactPosition", "contact_position"),
                ("WMSFees", "fees"),
                ("WMSKeywordList", "keyword_list"),
                ("WMSOnlineResource", "online_resource"),
                ("WMSServiceAbstract", "service_abstract"),
                ("WMSServiceTitle", "service_title"),
                ("WMSUrl", "resource_url"),
            ],
            "keys": ["/"],
        }
    }

    scopes = (
        [scope_or_scopes] if isinstance(scope_or_scopes, str) else scope_or_scopes
    )

    for scope in scopes:
        if scope not in supported_scopes:
            supported = ", ".join(supported_scopes.keys())
            error_detail = f"This scope is not supported: {scope}. Supported scopes: {supported}"
            raise ValueError(error_detail)

        scope_entries = supported_scopes[scope]["scopes"]
        key_entries = supported_scopes[scope]["keys"]
        to_collect = zip_longest(
            scope_entries, key_entries, fillvalue=key_entries[0]
        )

        def collect(acc, pair):
            scope, key = pair
            qgis_scope_name, our_scope_name = scope

            if "list" in qgis_scope_name.lower():
                # PyQGIS sometimes violates Liskov's substitution principle so naming tricks needed
                list_as_text = ", ".join(
                    project.readListEntry(qgis_scope_name, key)[0]
                )
                acc.append((our_scope_name, list_as_text))
            else:
                acc.append(
                    (our_scope_name, project.readEntry(qgis_scope_name, key)[0])
                )

            return acc

        return dict(reduce(collect, to_collect, []))

make_wgs84_geom_transform(project, layer) -> QgsCoordinateTransform staticmethod

Creates a QgisCoordinateTransform context to transform a job_layer_definition to EPSG:4326 aka wgs84.

Parameters:

• project –

  The QGIS project instance.

• layer –

  The job_layer_definition which's extent should be reprojected.

Returns:

• QgsCoordinateTransform –

  The QGIS transformation context.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def make_wgs84_geom_transform(project, layer) -> QgsCoordinateTransform:
    """
    Creates a QgisCoordinateTransform context to transform a job_layer_definition to EPSG:4326 aka wgs84.

    Args:
        project: The QGIS project instance.
        layer: The job_layer_definition which's extent should be reprojected.

    Returns:
        The QGIS transformation context.
    """
    source_crs = layer.crs()
    epsg_4326 = QgsCoordinateReferenceSystem("EPSG:4326")
    return QgsCoordinateTransform(source_crs, epsg_4326, project)

merge_dicts(a, b) staticmethod

Merges two dicts recursively, b values overwrites a values.

Parameters:

• a –

  Dictionary which is the base.

• b –

  Dictionary which is merged in and whose values overwrites the one.

Returns:

• –

  The merged dict.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def merge_dicts(a, b):
    """
    Merges two dicts recursively, b values overwrites a values.

    Args:
        a: Dictionary which is the base.
        b: Dictionary which is merged in and whose values overwrites the one.

    Returns:
        The merged dict.
    """
    result = a.copy()
    for key, value in b.items():
        if (
            key in result
            and isinstance(result[key], dict)
            and isinstance(value, dict)
        ):
            result[key] = Exporter.merge_dicts(result[key], value)
        else:
            result[key] = value
    return result

obtain_editor_widget_type_from_qgis_field(field: QgsField) -> Tuple[str, str, str, str] | Tuple[str, None, None, None] staticmethod

We simply mimikri QGIS Server here

TODO: This could be improved alot! Maybe we can also backport that to QGIS core some day?

Parameters:

• field (QgsField) –

  The field of an QgsVectorLayer.

Returns:

• Tuple[str, str, str, str] | Tuple[str, None, None, None] –

  Unified type name regarding

• Tuple[str, str, str, str] | Tuple[str, None, None, None] –

  XSD spec

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def obtain_editor_widget_type_from_qgis_field(
    field: QgsField,
) -> Tuple[str, str, str, str] | Tuple[str, None, None, None]:
    """
    We simply mimikri [QGIS Server here](https://github.com/qgis/QGIS/blob/de98779ebb117547364ec4cff433f062374e84a3/src/server/services/wfs/qgswfsdescribefeaturetype.cpp#L153-L192)

    TODO: This could be improved alot! Maybe we can also backport that to QGIS core some day?

    Args:
        field: The field of an `QgsVectorLayer`.

    Returns:
        Unified type name regarding
        [XSD spec](https://www.w3.org/TR/xmlschema11-2/#built-in-primitive-datatypes)
    """
    attribute_type = field.type()
    setup = field.editorWidgetSetup()
    config = setup.config()
    editor_widget_type = setup.type()
    if editor_widget_type == "DateTime":
        field_format = config.get(
            "field_format", QgsDateTimeFieldFormatter.defaultFormat(attribute_type)
        )
        if field_format == QgsDateTimeFieldFormatter.TIME_FORMAT:
            return editor_widget_type, "time", "string", "time"
        elif field_format == QgsDateTimeFieldFormatter.DATE_FORMAT:
            return editor_widget_type, "date", "string", "date"
        elif field_format == QgsDateTimeFieldFormatter.DATETIME_FORMAT:
            return editor_widget_type, "dateTime", "string", "date-time"
        elif field_format == QgsDateTimeFieldFormatter.QT_ISO_FORMAT:
            return editor_widget_type, "dateTime", "string", "date-time"
    elif editor_widget_type == "Range":
        if config.get("Precision"):
            config_precision = int(config["Precision"])
            if config_precision != field.precision():
                if config_precision == 0:
                    return editor_widget_type, "integer", "integer", "int64"
                else:
                    return editor_widget_type, "decimal", "number", "double"

    logging.error(
        f"Editor widget type was not handled as expected: {editor_widget_type}"
    )
    return editor_widget_type, None, None, None

obtain_nullable(field: QgsField) staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def obtain_nullable(field: QgsField):
    if not (
        field.constraints().constraints()
        == QgsFieldConstraints.Constraint.ConstraintNotNull
    ):
        return True
    return False

obtain_simple_types_from_qgis_field_json(field: QgsField) -> Tuple[str, str] | Tuple[str, None] staticmethod

Delivers the type match for json based on the field QgsField type.

Parameters:

• field (QgsField) –

  The field of an QgsVectorLayer.

Returns:

• Tuple[str, str] | Tuple[str, None] –

  Unified type name and format in a tuple.

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def obtain_simple_types_from_qgis_field_json(
    field: QgsField,
) -> Tuple[str, str] | Tuple[str, None]:
    """
    Delivers the type match for json based on the field QgsField type.

    Args:
        field: The field of an `QgsVectorLayer`.

    Returns:
        Unified type name and format in a tuple.
    """
    attribute_type = field.type()
    if attribute_type == QMetaType.Type.Int:
        return "integer", None
    elif attribute_type == QMetaType.Type.UInt:
        return "integer", "uint32"
    elif attribute_type == QMetaType.Type.LongLong:
        return "integer", "int64"
    elif attribute_type == QMetaType.Type.ULongLong:
        return "integer", "uint64"
    elif attribute_type == QMetaType.Type.Double:
        return "number", "double"
    elif attribute_type == QMetaType.Type.Float:
        return "number", "float"
    elif attribute_type == QMetaType.Type.Bool:
        return "boolean", None
    elif attribute_type == QMetaType.Type.QDate:
        return "string", "date"
    elif attribute_type == QMetaType.Type.QTime:
        return "string", "time"
    elif attribute_type == QMetaType.Type.QDateTime:
        return "string", "date-time"
    else:
        # we handle all unknown types as string. Since its for JavaScript, this should be safe.
        return "string", None

obtain_simple_types_from_qgis_field_xml(field: QgsField) -> str staticmethod

Parameters:

• field (QgsField) –

  The field of an QgsVectorLayer.

Returns:

• str –

  Unified type name regarding

• str –

  XSD spec

• IMPORTANT ( str ) –

  If type is not matched within the function it will be string always!

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def obtain_simple_types_from_qgis_field_xml(field: QgsField) -> str:
    """

    Args:
        field: The field of an `QgsVectorLayer`.

    Returns:
        Unified type name regarding
        [XSD spec](https://www.w3.org/TR/xmlschema11-2/#built-in-primitive-datatypes)
        IMPORTANT: If type is not matched within the function it will be `string` always!
    """
    attribute_type = field.type()
    if attribute_type == QMetaType.Type.Int:
        return "int"
    elif attribute_type == QMetaType.Type.UInt:
        return "unsignedInt"
    elif attribute_type == QMetaType.Type.LongLong:
        return "long"
    elif attribute_type == QMetaType.Type.ULongLong:
        return "unsignedLong"
    elif attribute_type == QMetaType.Type.Double:
        if field.length() > 0 and field.precision() == 0:
            return "integer"
        else:
            return "decimal"
    elif attribute_type == QMetaType.Type.Bool:
        return "boolean"
    elif attribute_type == QMetaType.Type.QDate:
        return "date"
    elif attribute_type == QMetaType.Type.QTime:
        return "time"
    elif attribute_type == QMetaType.Type.QDateTime:
        return "dateTime"
    else:
        return "string"

open_qgis_project(path_to_project: str) -> QgsProject staticmethod

Parameters:

• path_to_project (str) –

  The absolute path on the file system where the project can be read from.

Returns:

• QgsProject –

  The opened project (read).

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def open_qgis_project(path_to_project: str) -> QgsProject:
    """


    Args:
        path_to_project: The absolute path on the file system where the project can be read from.

    Returns:
        The opened project (read).
    """
    project = QgsProject.instance()
    project.read(path_to_project)
    return project

prepare_qgis_project_name(project: QgsProject) -> tuple[str, str] staticmethod

Parameters:

• project (QgsProject) –

  The instantiated QGIS project.

Returns:

• tuple[str, str] –

  Tuple of str version, name

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def prepare_qgis_project_name(project: QgsProject) -> tuple[str, str]:
    """


    Args:
        project: The instantiated QGIS project.

    Returns:
        Tuple of str version, name
    """
    # TODO: Find a good approach to recognize different "versions" of a project.
    name = project.baseName()
    parts = name.split(".")
    version = parts.pop(0)
    assembled_name = ".".join(parts)
    if assembled_name == "":
        assembled_name = project.title()
    return version, assembled_name

provide_field_length(field: QgsField) -> int | None staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def provide_field_length(field: QgsField) -> int | None:
    length = field.length()
    if length > 0:
        return length
    else:
        return None

provide_field_precision(field: QgsField) -> int | None staticmethod

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def provide_field_precision(field: QgsField) -> int | None:
    precision = field.precision()
    if precision > 0:
        return precision
    else:
        return None

run() -> Config

Source code in src/qgis_server_light/exporter/extract.py

def run(self) -> Config:
    self.walk_qgis_project_tree(self.qgis_project_tree_root, [])
    return self.qsl_config

sanitize_name(raw: str, lower: bool = False) -> str staticmethod

Transforms an arbitrary string into a WMS/WFS and URL compatible short name for a job_layer_definition or group.

Steps: 1. Unicode‑NFD → ASCII‑transliteration (removes umlauts/diacritics). 2. All chars, which are NOT [A‑Za‑z0‑9_.‑], will be replaced by '' ersetzen. 3. Reduce multiple underscore '' to a single one. 4. Remove leading '', '.', '-'. 5. If the string is empty OR does not start with a letter OR not start with '', a leading '_' will be added. 6. Optional all will be converted to lowercase (lower=True).

Source code in src/qgis_server_light/exporter/extract.py

@staticmethod
def sanitize_name(raw: str, lower: bool = False) -> str:
    """
    Transforms an arbitrary string into a WMS/WFS and URL compatible short name for a job_layer_definition or group.

    Steps:
    1. Unicode‑NFD → ASCII‑transliteration (removes umlauts/diacritics).
    2. All chars, which are NOT [A‑Za‑z0‑9_.‑], will be replaced by '_' ersetzen.
    3. Reduce multiple underscore '_' to a single one.
    4. Remove leading '_', '.', '-'.
    5. If the string is empty OR does not start with a letter OR not start with '_',
       a leading '_' will be added.
    6. Optional all will be converted to lowercase (lower=True).
    """
    # 1. cleaning to ASCII
    ascii_str = (
        unicodedata.normalize("NFKD", raw).encode("ascii", "ignore").decode()
    )
    # 2. not allowed → '_'
    ascii_str = re.sub(r"[^A-Za-z0-9_.-]+", "_", ascii_str)
    # 3. remove multiple '_'
    ascii_str = re.sub(r"_+", "_", ascii_str)
    # 4. remove trailing chars
    ascii_str = ascii_str.strip("._-")
    # 5. ensure first char is correct (mainly xml stuff and URL)
    if not ascii_str or not re.match(r"[A-Za-z_]", ascii_str[0]):
        ascii_str = "_" + ascii_str
    # 6. Optional lowercase
    if lower:
        ascii_str = ascii_str.lower()
    return ascii_str

short_name(short_name: str, path: list[str]) -> str

Decides if to use the short name itself or the unified version by the tree path.

Parameters:

• short_name (str) –

  The short name either of the group or the job_layer_definition.

• path (list[str]) –

  The path is a list of string which stores the information of the current tree path. This is used to construct a string for unifying job_layer_definition names by their tree path.

Returns:

• str –

  The short name itself or its unified tree path.

Source code in src/qgis_server_light/exporter/extract.py

def short_name(self, short_name: str, path: list[str]) -> str:
    """
    Decides if to use the short name itself or the unified version by the tree path.

    Args:
        short_name: The short name either of the group or the job_layer_definition.
        path: The path is a list of string which stores the information of the current tree path. This is
            used to construct a string for unifying job_layer_definition names by their tree path.

    Returns:
        The short name itself or its unified tree path.
    """
    if self.unify_layer_names_by_group:
        return self.create_unified_short_name(short_name, path)
    else:
        return short_name

walk_qgis_project_tree(entity: QgsLayerTreeNode, path: list[str])

This is a highly recursive function which walks to the qgis job_layer_definition tree to extract all knowledge out of it. It is called from itself again for each level of group like elements which are found.

Parameters:

• entity (QgsLayerTreeNode) –

  The QGIS projects tree node which can be a QgsLayerTree, QgsLayerTreeGroup or QgsLayerTreeLayer.

• path (list[str]) –

  The path is a list of string which stores the information of the current tree path. This is used to construct a string for unifying job_layer_definition names by their tree path.

Source code in src/qgis_server_light/exporter/extract.py

def walk_qgis_project_tree(
    self,
    entity: QgsLayerTreeNode,
    path: list[str],
):
    """
    This is a highly recursive function which walks to the qgis job_layer_definition tree to extract all knowledge out
    of it. It is called from itself again for each level of group like elements which are found.

    Args:
        entity: The QGIS projects tree node which can be a QgsLayerTree, QgsLayerTreeGroup or
            QgsLayerTreeLayer.
        path: The path is a list of string which stores the information of the current tree path. This is
            used to construct a string for unifying job_layer_definition names by their tree path.
    """
    if isinstance(entity, QgsLayerTreeLayer):
        # this is a job_layer_definition, we can extract its information
        self.extract_save_layer(
            entity,
            path,
        )
    elif isinstance(entity, QgsLayerTreeGroup) or isinstance(entity, QgsLayerTree):
        # these are "group like" elements, we need to step into them one level deeper.
        short_name = self.get_group_short_name(entity)
        if short_name != "":
            # '' is the root of the tree, we don't want it to be part of the path
            path = path + [short_name]
        self.extract_group(
            entity,
            path,
        )
        for child in entity.children():
            # we handle all the children of the group like thing.
            self.walk_qgis_project_tree(
                child,
                path,
            )
    else:
        logging.error(
            f"This element was not expected while walking QGIS project tree: {entity}"
        )

interface

common

This module contains common logic, shared beyond all specialized parts of the QGIS-Server-Light interface.

BBox dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/common.py

@dataclass(repr=False)
class BBox(BaseInterface):
    x_min: float = field(metadata={"type": "Element"})
    x_max: float = field(metadata={"type": "Element"})
    y_min: float = field(metadata={"type": "Element"})
    y_max: float = field(metadata={"type": "Element"})
    z_min: float = field(default=0.0, metadata={"type": "Element"})
    z_max: float = field(default=0.0, metadata={"type": "Element"})

    def to_list(self) -> list:
        return [self.x_min, self.y_min, self.z_min, self.x_max, self.y_max, self.z_max]

    def to_string(self) -> str:
        return ",".join([str(item) for item in self.to_list()])

    def to_2d_list(self) -> list:
        return [self.x_min, self.y_min, self.x_max, self.y_max]

    def to_2d_string(self) -> str:
        return ",".join([str(item) for item in self.to_2d_list()])

    @staticmethod
    def from_string(bbox_string: str) -> "BBox":
        """
        Takes a CSV string representation of a BBox in the form:
            '<x_min>,<y_min>,<x_max>,<y_max>' or
            '<x_min>,<y_min>,<z_min>,<x_max>,<y_max>,<z_max>'
        """
        coordinates = bbox_string.split(",")
        if len(coordinates) == 4:
            return BBox(
                x_min=float(coordinates[0]),
                y_min=float(coordinates[1]),
                x_max=float(coordinates[2]),
                y_max=float(coordinates[3]),
            )
        elif len(coordinates) == 6:
            return BBox(
                x_min=float(coordinates[0]),
                y_min=float(coordinates[1]),
                z_min=float(coordinates[2]),
                x_max=float(coordinates[3]),
                y_max=float(coordinates[4]),
                z_max=float(coordinates[5]),
            )
        else:
            raise ValueError(f"Invalid bbox string: {bbox_string}")

    @staticmethod
    def from_list(bbox_list: List[float]) -> "BBox":
        """
        Takes a list representation of a BBox in the form:
            [<x_min>,<y_min>,<x_max>,<y_max>] or
            [<x_min>,<y_min>,<z_min>,<x_max>,<y_max>,<z_max>]
        """
        if len(bbox_list) == 4:
            return BBox(
                x_min=bbox_list[0],
                y_min=bbox_list[1],
                x_max=bbox_list[2],
                y_max=bbox_list[3],
            )
        elif len(bbox_list) == 6:
            return BBox(
                x_min=bbox_list[0],
                y_min=bbox_list[1],
                z_min=bbox_list[2],
                x_max=bbox_list[3],
                y_max=bbox_list[4],
                z_max=bbox_list[5],
            )
        else:
            raise ValueError(f"Invalid bbox list: {bbox_list}")

x_max: float = field(metadata={'type': 'Element'}) class-attribute instance-attribute

x_min: float = field(metadata={'type': 'Element'}) class-attribute instance-attribute

y_max: float = field(metadata={'type': 'Element'}) class-attribute instance-attribute

y_min: float = field(metadata={'type': 'Element'}) class-attribute instance-attribute

z_max: float = field(default=0.0, metadata={'type': 'Element'}) class-attribute instance-attribute

z_min: float = field(default=0.0, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(x_min: float, x_max: float, y_min: float, y_max: float, z_min: float = 0.0, z_max: float = 0.0) -> None

from_list(bbox_list: List[float]) -> BBox staticmethod

Takes a list representation of a BBox in the form

[,,,] or [,,,,,]

Source code in src/qgis_server_light/interface/common.py

@staticmethod
def from_list(bbox_list: List[float]) -> "BBox":
    """
    Takes a list representation of a BBox in the form:
        [<x_min>,<y_min>,<x_max>,<y_max>] or
        [<x_min>,<y_min>,<z_min>,<x_max>,<y_max>,<z_max>]
    """
    if len(bbox_list) == 4:
        return BBox(
            x_min=bbox_list[0],
            y_min=bbox_list[1],
            x_max=bbox_list[2],
            y_max=bbox_list[3],
        )
    elif len(bbox_list) == 6:
        return BBox(
            x_min=bbox_list[0],
            y_min=bbox_list[1],
            z_min=bbox_list[2],
            x_max=bbox_list[3],
            y_max=bbox_list[4],
            z_max=bbox_list[5],
        )
    else:
        raise ValueError(f"Invalid bbox list: {bbox_list}")

from_string(bbox_string: str) -> BBox staticmethod

Takes a CSV string representation of a BBox in the form

',,,' or ',,,,,'

Source code in src/qgis_server_light/interface/common.py

@staticmethod
def from_string(bbox_string: str) -> "BBox":
    """
    Takes a CSV string representation of a BBox in the form:
        '<x_min>,<y_min>,<x_max>,<y_max>' or
        '<x_min>,<y_min>,<z_min>,<x_max>,<y_max>,<z_max>'
    """
    coordinates = bbox_string.split(",")
    if len(coordinates) == 4:
        return BBox(
            x_min=float(coordinates[0]),
            y_min=float(coordinates[1]),
            x_max=float(coordinates[2]),
            y_max=float(coordinates[3]),
        )
    elif len(coordinates) == 6:
        return BBox(
            x_min=float(coordinates[0]),
            y_min=float(coordinates[1]),
            z_min=float(coordinates[2]),
            x_max=float(coordinates[3]),
            y_max=float(coordinates[4]),
            z_max=float(coordinates[5]),
        )
    else:
        raise ValueError(f"Invalid bbox string: {bbox_string}")

to_2d_list() -> list

Source code in src/qgis_server_light/interface/common.py

def to_2d_list(self) -> list:
    return [self.x_min, self.y_min, self.x_max, self.y_max]

to_2d_string() -> str

Source code in src/qgis_server_light/interface/common.py

def to_2d_string(self) -> str:
    return ",".join([str(item) for item in self.to_2d_list()])

to_list() -> list

Source code in src/qgis_server_light/interface/common.py

def to_list(self) -> list:
    return [self.x_min, self.y_min, self.z_min, self.x_max, self.y_max, self.z_max]

to_string() -> str

Source code in src/qgis_server_light/interface/common.py

def to_string(self) -> str:
    return ",".join([str(item) for item in self.to_list()])

BaseInterface dataclass

This class should be used as base class for all dataclasses in the interface. It offers useful methods to handle exposed content in a centralized way. Mainly for logging redaction.

Since dataclasses gets a repr method installed automatically when they are created, a dataclass inheriting from this base class has to be defined as follows:

@dataclass(repr=False)
class Config(BaseInterface):
    id: int = field(metadata={"type": "Element"})
    secure: str = field(metadata={"type": "Element"})
    long_content: str = field(metadata={"type": "Element"})

    @property
    def shortened_fields(self) -> set:
        return {"long_content"}

    @property
    def redacted_fields(self) -> set:
        return {"secure"}

This way, when an instance of this example class gets logged somewhere it the output will be redacted, meaning the logging output might look like this:

Config(id=1, secure=**REDACTED**, long_content=abc12...io345)

Source code in src/qgis_server_light/interface/common.py

@dataclass
class BaseInterface:
    """
    This class should be used as base class for all dataclasses in the interface. It offers useful methods to
    handle exposed content in a centralized way. Mainly for logging redaction.

    Since dataclasses gets a __repr__ method installed automatically when they are created, a dataclass
    inheriting from this base class has to be defined as follows:

        @dataclass(repr=False)
        class Config(BaseInterface):
            id: int = field(metadata={"type": "Element"})
            secure: str = field(metadata={"type": "Element"})
            long_content: str = field(metadata={"type": "Element"})

            @property
            def shortened_fields(self) -> set:
                return {"long_content"}

            @property
            def redacted_fields(self) -> set:
                return {"secure"}

    This way, when an instance of this example class gets logged somewhere it the output will be redacted,
    meaning the logging output might look like this:

        Config(id=1, secure=**REDACTED**, long_content=abc12...io345)

    """

    @property
    def shorten_limit(self) -> int:
        """
        The limit to which the content of a field should be shortened.

        Returns:
            The limit.
        """
        return 5

    @property
    def redacted_fields(self) -> set:
        """
        Field which contents should get redacted before printing them on the log. This is mainly used to
        prevent passwords in logs.

        Returns:
            The set of field names which should be redacted
        """
        return set()

    @property
    def shortened_fields(self) -> set:
        """
        Fields which should be shortened to a length, this is manly useful for large content fields with
        BLOB etc.

        Returns:
            The set field names which should be shortened.
        """
        return set()

    def _value_string(self, repr_value: str | bytes):
        return f"{repr_value[: self.shorten_limit]}...{repr_value[((1 + self.shorten_limit) * -1) :]}"

    def _type_aware_value_string(self, value, repr_value):
        value_string = self._value_string(repr_value)
        if type(value) in [str]:
            return f"'{value_string}'"
        else:
            return f"{value_string}"

    def __repr__(self):
        members = []
        cls = self.__class__.__name__
        for obj_field in fields(self):
            # this is the original switch dataclasses allow on fields
            if obj_field.repr:
                value = getattr(self, obj_field.name)
                repr_value = str(value)
                if obj_field.name in self.redacted_fields:
                    members.append(f"{obj_field.name}=**REDACTED**")
                elif (
                    obj_field.name in self.shortened_fields
                    and value is not None
                    and len(repr_value) > self.shorten_limit * 2
                ):
                    members.append(
                        f"{obj_field.name}={self._type_aware_value_string(value, repr_value)}"
                    )
                else:
                    members.append(f"{obj_field.name}={value!r}")
        return f"{cls}({', '.join(members)})"

redacted_fields: set property

Field which contents should get redacted before printing them on the log. This is mainly used to prevent passwords in logs.

Returns:

• set –

  The set of field names which should be redacted

shorten_limit: int property

The limit to which the content of a field should be shortened.

Returns:

• int –

  The limit.

shortened_fields: set property

Fields which should be shortened to a length, this is manly useful for large content fields with BLOB etc.

Returns:

• set –

  The set field names which should be shortened.

__init__() -> None

__repr__()

Source code in src/qgis_server_light/interface/common.py

def __repr__(self):
    members = []
    cls = self.__class__.__name__
    for obj_field in fields(self):
        # this is the original switch dataclasses allow on fields
        if obj_field.repr:
            value = getattr(self, obj_field.name)
            repr_value = str(value)
            if obj_field.name in self.redacted_fields:
                members.append(f"{obj_field.name}=**REDACTED**")
            elif (
                obj_field.name in self.shortened_fields
                and value is not None
                and len(repr_value) > self.shorten_limit * 2
            ):
                members.append(
                    f"{obj_field.name}={self._type_aware_value_string(value, repr_value)}"
                )
            else:
                members.append(f"{obj_field.name}={value!r}")
    return f"{cls}({', '.join(members)})"

PgServiceConf dataclass

Bases: BaseInterface

A typed definition of the pg_service.conf definition which might be used.

Source code in src/qgis_server_light/interface/common.py

@dataclass(repr=False)
class PgServiceConf(BaseInterface):
    """
    A typed definition of the pg_service.conf definition which might be used.
    """

    name: str = field(metadata={"type": "Element"})
    host: str | None = field(
        default=None,
        metadata={"type": "Element"},
    )
    port: int | None = field(default=None, metadata={"type": "Element"})
    user: str | None = field(default=None, metadata={"type": "Element"})
    dbname: str | None = field(default=None, metadata={"type": "Element"})
    password: str | None = field(default=None, metadata={"type": "Element"})
    sslmode: SslMode = field(default=SslMode.PREFER, metadata={"type": "Element"})
    application_name: str | None = field(default=None, metadata={"type": "Element"})
    client_encoding: str = field(default="UTF8", metadata={"type": "Element"})
    # possibilitiy to link to another service (nested definitions!)
    service: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def redacted_fields(self) -> set:
        return {"password"}

application_name: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

client_encoding: str = field(default='UTF8', metadata={'type': 'Element'}) class-attribute instance-attribute

dbname: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

host: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

password: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

port: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

redacted_fields: set property

service: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

sslmode: SslMode = field(default=(SslMode.PREFER), metadata={'type': 'Element'}) class-attribute instance-attribute

user: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(name: str, host: str | None = None, port: int | None = None, user: str | None = None, dbname: str | None = None, password: str | None = None, sslmode: SslMode = SslMode.PREFER, application_name: str | None = None, client_encoding: str = 'UTF8', service: str | None = None) -> None

RedactedString

This special string class can be used to handle secret strings in the application. It works like a normal string but in case it's used to print or log its value is not reveled to the output.

Source code in src/qgis_server_light/interface/common.py

class RedactedString:
    """
    This special string class can be used to handle secret strings in the application. It works like a normal
    string but in case it's used to print or log its value is not reveled to the output.
    """

    def __init__(self, value, redacted_text="**REDACTED**"):
        self._value = value
        self._redacted_text = redacted_text

    def __str__(self):
        return self._redacted_text

    def __repr__(self):
        return f"<RedactedString {self._redacted_text}>"

    def __format__(self, format_spec):
        return self._redacted_text

    def __json__(self):
        return self._redacted_text

    def reveal(self):
        """
        Allows access to the original value when necessary.

        Returns:
            The secret string.
        """

        return self._value

__format__(format_spec)

Source code in src/qgis_server_light/interface/common.py

def __format__(self, format_spec):
    return self._redacted_text

__init__(value, redacted_text='**REDACTED**')

Source code in src/qgis_server_light/interface/common.py

def __init__(self, value, redacted_text="**REDACTED**"):
    self._value = value
    self._redacted_text = redacted_text

__json__()

Source code in src/qgis_server_light/interface/common.py

def __json__(self):
    return self._redacted_text

__repr__()

Source code in src/qgis_server_light/interface/common.py

def __repr__(self):
    return f"<RedactedString {self._redacted_text}>"

__str__()

Source code in src/qgis_server_light/interface/common.py

def __str__(self):
    return self._redacted_text

reveal()

Allows access to the original value when necessary.

Returns:

• –

  The secret string.

Source code in src/qgis_server_light/interface/common.py

def reveal(self):
    """
    Allows access to the original value when necessary.

    Returns:
        The secret string.
    """

    return self._value

SslMode

Bases: str, Enum

Source code in src/qgis_server_light/interface/common.py

class SslMode(str, Enum):
    DISABLE = "disable"
    ALLOW = "allow"
    PREFER = "prefer"
    REQUIRE = "require"
    VERIFY_CA = "verify-ca"
    VERIFY_FULL = "verify-full"

ALLOW = 'allow' class-attribute instance-attribute

DISABLE = 'disable' class-attribute instance-attribute

PREFER = 'prefer' class-attribute instance-attribute

REQUIRE = 'require' class-attribute instance-attribute

VERIFY_CA = 'verify-ca' class-attribute instance-attribute

VERIFY_FULL = 'verify-full' class-attribute instance-attribute

Style dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/common.py

@dataclass(repr=False)
class Style(BaseInterface):
    name: str = field(metadata={"type": "Element"})
    definition: str = field(metadata={"type": "Element"})

    @property
    def shortened_fields(self) -> set:
        return {"definition"}

definition: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

shortened_fields: set property

__init__(name: str, definition: str) -> None

dispatcher

common

Status

Bases: Enum

Source code in src/qgis_server_light/interface/dispatcher/common.py

class Status(Enum):
    SUCCESS = "succeed"
    FAILURE = "failed"
    RUNNING = "running"
    QUEUED = "queued"

FAILURE = 'failed' class-attribute instance-attribute

QUEUED = 'queued' class-attribute instance-attribute

RUNNING = 'running' class-attribute instance-attribute

SUCCESS = 'succeed' class-attribute instance-attribute

redis_asio

This contains the interface definition about how a job info is passed around a redis queue.

RedisQueue

Source code in src/qgis_server_light/interface/dispatcher/redis_asio.py

class RedisQueue:
    job_queue_name: str = "jobs"
    job_info_key: str = "info"
    job_info_type_key: str = "info_type"
    job_channel_name: str = "notifications"
    job_status_key: str = "status"
    job_duration_key: str = "duration"
    job_timestamp_key: str = "timestamp"
    job_last_update_key: str = f"{job_timestamp_key}.last_update"

    def __init__(self, redis_client: redis_aio.Redis) -> None:
        # we use this to hold connections to redis in a pool, this way we are
        # event loop safe and when creating the redis client for every call of
        # post, we only instantiate a minimal wrapper object which is cheap.

        self.client = redis_client

    @classmethod
    def create(cls, url: str):
        redis_client = redis_aio.Redis.from_url(url)
        return cls(redis_client)

    async def set_job_runtime_status(
        self,
        job_id,
        pipeline: Pipeline,
        status: str,
        start_time: float,
    ):
        duration = time.time() - start_time
        ts = datetime.datetime.now().isoformat()
        await pipeline.hset(f"job:{job_id}", self.job_status_key, status)
        await pipeline.hset(
            f"job:{job_id}",
            f"{self.job_timestamp_key}.{status}",
            ts,
        )
        await pipeline.hset(f"job:{job_id}", self.job_last_update_key, ts)
        await pipeline.hset(f"job:{job_id}", self.job_duration_key, str(duration))
        await pipeline.execute()

    async def post(
        self,
        job_parameter: (
            QslJobParameterRender
            | QslJobParameterFeatureInfo
            | QslJobParameterLegend
            | QslJobParameterFeature
        ),
        to: float = 10.0,
    ) -> tuple[JobResult, str]:
        """
        Posts a new `runner` to the runner queue and waits maximum `timeout` seconds to complete.
        Will return a JobResult if successful or raise an error.

        Args:
            job_parameter: The parameter for the job which should be executed.
            to: The timeout a job is expected to be waited for before canceling
                job execution.
        """
        job_id = str(uuid4())
        start_time = time.time()
        if isinstance(job_parameter, QslJobParameterRender):
            job_info = QslJobInfoRender(
                id=job_id, type=QslJobInfoRender.__name__, job=job_parameter
            )
        elif isinstance(job_parameter, QslJobParameterFeatureInfo):
            job_info = QslJobInfoFeatureInfo(
                id=job_id, type=QslJobParameterFeatureInfo.__name__, job=job_parameter
            )
        elif isinstance(job_parameter, QslJobParameterLegend):
            job_info = QslJobInfoLegend(
                id=job_id, type=QslJobInfoLegend.__name__, job=job_parameter
            )
        elif isinstance(job_parameter, QslJobParameterFeature):
            job_info = QslJobInfoFeature(
                id=job_id, type=QslJobInfoFeature.__name__, job=job_parameter
            )
        else:
            return (
                JobResult(
                    id=job_id,
                    data=f"Unsupported runner type: {type(job_parameter)}",
                    content_type="application/text",
                ),
                Status.FAILURE.value,
            )
        async with self.client.pipeline() as p:
            # Putting job info into redis
            await p.hset(
                f"job:{job_id}", self.job_info_key, JsonSerializer().render(job_info)
            )
            await p.hset(
                f"job:{job_id}", self.job_info_type_key, job_info.__class__.__name__
            )
            # Queuing the job onto the list/queue
            await p.rpush(self.job_queue_name, job_id)
            await p.execute()

            logging.info(f"{job_id} queued")

            # we inform, that the job was queued
            await self.set_job_runtime_status(
                job_id, p, Status.QUEUED.value, start_time
            )
            try:
                async with self.client.pubsub() as ps:
                    # we tell redis to let us know if a message is published
                    # for this channel `notifications:{job_id}`.
                    await ps.subscribe(f"{self.job_channel_name}:{job_id}")
                    try:
                        # this puts a timeout trigger on the subscription, after timeout
                        # an asyncio.TimeoutError or asyncio.exceptions.CancelledError
                        # is raised. See except block below.
                        async with timeout(to):
                            while True:
                                message = await ps.get_message(
                                    timeout=to, ignore_subscribe_messages=True
                                )
                                if not message:
                                    continue  # https://github.com/redis/redis-py/issues/733
                                status_binary = await self.client.hget(
                                    f"job:{job_id}", "status"
                                )
                                status = status_binary.decode()
                                result: JobResult = pickle.loads(message["data"])
                                duration = time.time() - start_time
                                if status == Status.SUCCESS.value:
                                    logging.info(
                                        f"Job id: {job_id}, status: {status}, "
                                        f"duration: {duration}"
                                    )
                                elif status == Status.FAILURE.value:
                                    logging.info(
                                        f"Job id: {job_id}, status: {status}, "
                                        f"duration: {duration}, error: {result.data}"
                                    )
                                return result, status
                    except (asyncio.TimeoutError, asyncio.exceptions.CancelledError):
                        logging.info(f"{job_id} timeout")
                        raise
            except Exception as e:
                duration = time.time() - start_time
                logging.info(
                    f"Job id: {job_id}, status: {Status.FAILURE.value}, duration: "
                    f"{duration}",
                    exc_info=True,
                )
                return (
                    JobResult(
                        id=job_id,
                        data=str(e),
                        content_type="application/text",
                    ),
                    Status.FAILURE.value,
                )
            finally:
                try:
                    await self.client.delete(f"job:{job_id}")
                except Exception:
                    logging.warning(
                        f"Cleanup failed for {job_id}",
                        exc_info=True,
                    )

client = redis_client instance-attribute

job_channel_name: str = 'notifications' class-attribute instance-attribute

job_duration_key: str = 'duration' class-attribute instance-attribute

job_info_key: str = 'info' class-attribute instance-attribute

job_info_type_key: str = 'info_type' class-attribute instance-attribute

job_last_update_key: str = f'{job_timestamp_key}.last_update' class-attribute instance-attribute

job_queue_name: str = 'jobs' class-attribute instance-attribute

job_status_key: str = 'status' class-attribute instance-attribute

job_timestamp_key: str = 'timestamp' class-attribute instance-attribute

__init__(redis_client: redis_aio.Redis) -> None

Source code in src/qgis_server_light/interface/dispatcher/redis_asio.py

def __init__(self, redis_client: redis_aio.Redis) -> None:
    # we use this to hold connections to redis in a pool, this way we are
    # event loop safe and when creating the redis client for every call of
    # post, we only instantiate a minimal wrapper object which is cheap.

    self.client = redis_client

create(url: str) classmethod

Source code in src/qgis_server_light/interface/dispatcher/redis_asio.py

@classmethod
def create(cls, url: str):
    redis_client = redis_aio.Redis.from_url(url)
    return cls(redis_client)

post(job_parameter: QslJobParameterRender | QslJobParameterFeatureInfo | QslJobParameterLegend | QslJobParameterFeature, to: float = 10.0) -> tuple[JobResult, str] async

Posts a new runner to the runner queue and waits maximum timeout seconds to complete. Will return a JobResult if successful or raise an error.

Parameters:

• job_parameter (QslJobParameterRender | QslJobParameterFeatureInfo | QslJobParameterLegend | QslJobParameterFeature) –

  The parameter for the job which should be executed.

• to (float, default: 10.0 ) –

  The timeout a job is expected to be waited for before canceling job execution.

Source code in src/qgis_server_light/interface/dispatcher/redis_asio.py

async def post(
    self,
    job_parameter: (
        QslJobParameterRender
        | QslJobParameterFeatureInfo
        | QslJobParameterLegend
        | QslJobParameterFeature
    ),
    to: float = 10.0,
) -> tuple[JobResult, str]:
    """
    Posts a new `runner` to the runner queue and waits maximum `timeout` seconds to complete.
    Will return a JobResult if successful or raise an error.

    Args:
        job_parameter: The parameter for the job which should be executed.
        to: The timeout a job is expected to be waited for before canceling
            job execution.
    """
    job_id = str(uuid4())
    start_time = time.time()
    if isinstance(job_parameter, QslJobParameterRender):
        job_info = QslJobInfoRender(
            id=job_id, type=QslJobInfoRender.__name__, job=job_parameter
        )
    elif isinstance(job_parameter, QslJobParameterFeatureInfo):
        job_info = QslJobInfoFeatureInfo(
            id=job_id, type=QslJobParameterFeatureInfo.__name__, job=job_parameter
        )
    elif isinstance(job_parameter, QslJobParameterLegend):
        job_info = QslJobInfoLegend(
            id=job_id, type=QslJobInfoLegend.__name__, job=job_parameter
        )
    elif isinstance(job_parameter, QslJobParameterFeature):
        job_info = QslJobInfoFeature(
            id=job_id, type=QslJobInfoFeature.__name__, job=job_parameter
        )
    else:
        return (
            JobResult(
                id=job_id,
                data=f"Unsupported runner type: {type(job_parameter)}",
                content_type="application/text",
            ),
            Status.FAILURE.value,
        )
    async with self.client.pipeline() as p:
        # Putting job info into redis
        await p.hset(
            f"job:{job_id}", self.job_info_key, JsonSerializer().render(job_info)
        )
        await p.hset(
            f"job:{job_id}", self.job_info_type_key, job_info.__class__.__name__
        )
        # Queuing the job onto the list/queue
        await p.rpush(self.job_queue_name, job_id)
        await p.execute()

        logging.info(f"{job_id} queued")

        # we inform, that the job was queued
        await self.set_job_runtime_status(
            job_id, p, Status.QUEUED.value, start_time
        )
        try:
            async with self.client.pubsub() as ps:
                # we tell redis to let us know if a message is published
                # for this channel `notifications:{job_id}`.
                await ps.subscribe(f"{self.job_channel_name}:{job_id}")
                try:
                    # this puts a timeout trigger on the subscription, after timeout
                    # an asyncio.TimeoutError or asyncio.exceptions.CancelledError
                    # is raised. See except block below.
                    async with timeout(to):
                        while True:
                            message = await ps.get_message(
                                timeout=to, ignore_subscribe_messages=True
                            )
                            if not message:
                                continue  # https://github.com/redis/redis-py/issues/733
                            status_binary = await self.client.hget(
                                f"job:{job_id}", "status"
                            )
                            status = status_binary.decode()
                            result: JobResult = pickle.loads(message["data"])
                            duration = time.time() - start_time
                            if status == Status.SUCCESS.value:
                                logging.info(
                                    f"Job id: {job_id}, status: {status}, "
                                    f"duration: {duration}"
                                )
                            elif status == Status.FAILURE.value:
                                logging.info(
                                    f"Job id: {job_id}, status: {status}, "
                                    f"duration: {duration}, error: {result.data}"
                                )
                            return result, status
                except (asyncio.TimeoutError, asyncio.exceptions.CancelledError):
                    logging.info(f"{job_id} timeout")
                    raise
        except Exception as e:
            duration = time.time() - start_time
            logging.info(
                f"Job id: {job_id}, status: {Status.FAILURE.value}, duration: "
                f"{duration}",
                exc_info=True,
            )
            return (
                JobResult(
                    id=job_id,
                    data=str(e),
                    content_type="application/text",
                ),
                Status.FAILURE.value,
            )
        finally:
            try:
                await self.client.delete(f"job:{job_id}")
            except Exception:
                logging.warning(
                    f"Cleanup failed for {job_id}",
                    exc_info=True,
                )

set_job_runtime_status(job_id, pipeline: Pipeline, status: str, start_time: float) async

Source code in src/qgis_server_light/interface/dispatcher/redis_asio.py

async def set_job_runtime_status(
    self,
    job_id,
    pipeline: Pipeline,
    status: str,
    start_time: float,
):
    duration = time.time() - start_time
    ts = datetime.datetime.now().isoformat()
    await pipeline.hset(f"job:{job_id}", self.job_status_key, status)
    await pipeline.hset(
        f"job:{job_id}",
        f"{self.job_timestamp_key}.{status}",
        ts,
    )
    await pipeline.hset(f"job:{job_id}", self.job_last_update_key, ts)
    await pipeline.hset(f"job:{job_id}", self.job_duration_key, str(duration))
    await pipeline.execute()

exporter

api

ExportParameters dataclass

Bases: BaseInterface

The serializable request parameters which are accepted by the exporter service.

Source code in src/qgis_server_light/interface/exporter/api.py

@dataclass(repr=False)
class ExportParameters(BaseInterface):
    """
    The serializable request parameters which are accepted by the exporter service.
    """

    mandant: str = field(metadata={"type": "Element"})
    project: str = field(metadata={"type": "Element"})
    unify_layer_names_by_group: bool = field(
        metadata={"type": "Element"}, default=False
    )
    output_format: str = field(metadata={"type": "Element"}, default="json")
    pg_service_configs: list[PgServiceConf] = field(
        metadata={"type": "Element"}, default_factory=list
    )

    @property
    def pg_service_configs_dict(self) -> dict:
        configurations = {}
        for config in self.pg_service_configs:
            configurations[config.name] = {
                "host": config.host,
                "port": config.port,
                "user": config.user,
                "dbname": config.dbname,
                "password": config.password,
                "sslmode": config.sslmode,
                "application_name": config.application_name,
                "client_encoding": config.client_encoding,
                "service": config.service,
            }
        return configurations

mandant: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

output_format: str = field(metadata={'type': 'Element'}, default='json') class-attribute instance-attribute

pg_service_configs: list[PgServiceConf] = field(metadata={'type': 'Element'}, default_factory=list) class-attribute instance-attribute

pg_service_configs_dict: dict property

project: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

unify_layer_names_by_group: bool = field(metadata={'type': 'Element'}, default=False) class-attribute instance-attribute

__init__(mandant: str, project: str, unify_layer_names_by_group: bool = False, output_format: str = 'json', pg_service_configs: list[PgServiceConf] = list()) -> None

ExportResult dataclass

Bases: BaseInterface

The serializable response which is provided by the exporter service.

Source code in src/qgis_server_light/interface/exporter/api.py

@dataclass(repr=False)
class ExportResult(BaseInterface):
    """
    The serializable response which is provided by the exporter service.
    """

    successful: bool = field(metadata={"type": "Element"})

successful: bool = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(successful: bool) -> None

extract

This module contains all interface definition to translate from QGIS project to QGIS-Server-Light logic and to write the JSON export of the QGIS project

AbstractDataset dataclass

Bases: LayerLike

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class AbstractDataset(LayerLike):
    title: str = field(metadata={"type": "Element"})

title: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, name: str, title: str) -> None

Config dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Config(BaseInterface):
    project: Project = field(metadata={"type": "Element"})
    meta_data: MetaData = field(metadata={"type": "Element"})
    tree: Tree = field(metadata={"type": "Element"})
    datasets: Datasets = field(metadata={"type": "Element"})

datasets: Datasets = field(metadata={'type': 'Element'}) class-attribute instance-attribute

meta_data: MetaData = field(metadata={'type': 'Element'}) class-attribute instance-attribute

project: Project = field(metadata={'type': 'Element'}) class-attribute instance-attribute

tree: Tree = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(project: Project, meta_data: MetaData, tree: Tree, datasets: Datasets) -> None

Crs dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Crs(BaseInterface):
    auth_id: str = field(default=None, metadata={"type": "Element"})
    postgis_srid: int = field(
        default=None,
        metadata={"type": "Element"},
    )
    ogc_uri: str = field(default=None, metadata={"type": "Element"})
    ogc_urn: str = field(default=None, metadata={"type": "Element"})

auth_id: str = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

ogc_uri: str = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

ogc_urn: str = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

postgis_srid: int = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(auth_id: str = None, postgis_srid: int = None, ogc_uri: str = None, ogc_urn: str = None) -> None

Custom dataclass

Bases: DataSet

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Custom(DataSet):
    pass

__init__(id: str, name: str, title: str, source: DataSource, driver: str, bbox: BBox | None = None, bbox_wgs84: BBox | None = None, crs: Crs | None = None, styles: List[Style] = list(), minimum_scale: float | None = None, maximum_scale: float | None = None, style_name: str = 'default', is_spatial: bool = True) -> None

DataSet dataclass

Bases: AbstractDataset

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class DataSet(AbstractDataset):
    source: DataSource = field(metadata={"type": "Element"})
    driver: str = field(metadata={"type": "Element"})
    bbox: BBox | None = field(default=None, metadata={"type": "Element"})
    bbox_wgs84: BBox | None = field(default=None, metadata={"type": "Element"})
    crs: Crs | None = field(default=None, metadata={"type": "Element"})
    styles: List[Style] = field(default_factory=list, metadata={"type": "Element"})
    minimum_scale: float | None = field(default=None, metadata={"type": "Element"})
    maximum_scale: float | None = field(default=None, metadata={"type": "Element"})
    style_name: str = field(default="default", metadata={"type": "Element"})
    is_spatial: bool = field(default=True, metadata={"type": "Element"})

    def get_style_by_name(self, name: str) -> Style | None:
        for style in self.styles:
            if name == style.name:
                return style
        return None

    def style(self) -> Style | None:
        return self.get_style_by_name(self.style_name)

bbox: BBox | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

bbox_wgs84: BBox | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

crs: Crs | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

driver: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

is_spatial: bool = field(default=True, metadata={'type': 'Element'}) class-attribute instance-attribute

maximum_scale: float | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

minimum_scale: float | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

source: DataSource = field(metadata={'type': 'Element'}) class-attribute instance-attribute

style_name: str = field(default='default', metadata={'type': 'Element'}) class-attribute instance-attribute

styles: List[Style] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, name: str, title: str, source: DataSource, driver: str, bbox: BBox | None = None, bbox_wgs84: BBox | None = None, crs: Crs | None = None, styles: List[Style] = list(), minimum_scale: float | None = None, maximum_scale: float | None = None, style_name: str = 'default', is_spatial: bool = True) -> None

get_style_by_name(name: str) -> Style | None

Source code in src/qgis_server_light/interface/exporter/extract.py

def get_style_by_name(self, name: str) -> Style | None:
    for style in self.styles:
        if name == style.name:
            return style
    return None

style() -> Style | None

Source code in src/qgis_server_light/interface/exporter/extract.py

def style(self) -> Style | None:
    return self.get_style_by_name(self.style_name)

DataSource dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class DataSource(BaseInterface):
    postgres: PostgresSource | None = field(default=None, metadata={"type": "Element"})
    wmts: WmtsSource | None = field(default=None, metadata={"type": "Element"})
    wms: WmsSource | None = field(default=None, metadata={"type": "Element"})
    ogr: OgrSource | None = field(default=None, metadata={"type": "Element"})
    gdal: GdalSource | None = field(default=None, metadata={"type": "Element"})
    wfs: WfsSource | None = field(default=None, metadata={"type": "Element"})
    vector_tile: VectorTileSource | None = field(
        default=None, metadata={"type": "Element"}
    )
    xyz: XYZSource | None = field(default=None, metadata={"type": "Element"})

    @property
    def definition(
        self,
    ) -> (
        PostgresSource
        | WmtsSource
        | WmsSource
        | OgrSource
        | GdalSource
        | WfsSource
        | VectorTileSource
        | XYZSource
        | None
    ):
        for dataclass_field in fields(self):
            name = dataclass_field.name
            value = getattr(self, name)
            if value:
                return value
        logging.error(
            f"No source was definied at {self.__class__.__name__}, this is not expected"
        )
        return None

definition: PostgresSource | WmtsSource | WmsSource | OgrSource | GdalSource | WfsSource | VectorTileSource | XYZSource | None property

gdal: GdalSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

ogr: OgrSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

postgres: PostgresSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

vector_tile: VectorTileSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

wfs: WfsSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

wms: WmsSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

wmts: WmtsSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

xyz: XYZSource | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(postgres: PostgresSource | None = None, wmts: WmtsSource | None = None, wms: WmsSource | None = None, ogr: OgrSource | None = None, gdal: GdalSource | None = None, wfs: WfsSource | None = None, vector_tile: VectorTileSource | None = None, xyz: XYZSource | None = None) -> None

Datasets dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Datasets(BaseInterface):
    vector: list[Vector] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )
    raster: list[Raster] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )
    custom: list[Custom] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )
    group: list[Group] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )

custom: list[Custom] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

group: list[Group] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

raster: list[Raster] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

vector: list[Vector] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(vector: list[Vector] = list(), raster: list[Raster] = list(), custom: list[Custom] = list(), group: list[Group] = list()) -> None

Field dataclass

Bases: BaseInterface

Transportable (serializable) form of a QGIS vector job_layer_definition fiel (attribute). It contains the information of the original data datatype and its translated versions and the editor widget one as well.

Attributes:

• name (str) –

  Machine readable name of the field

• type (str) –

  Original type as defined by data source (PostGIS, GPKG, etc.)

• is_primary_key (bool) –

  if the field is considered to be primary key.

• type_wfs (Optional[str]) –

  Translated type for further usage. Based on the simple types of XSD spec.

• type_oapif (Optional[str]) –

  Translated type based on the types of the OpenAPI Spec

• type_oapif_format (Optional[str]) –

  Format of the above-mentioned type based on the OpenAPI Spec

• alias (Optional[str]) –

  Human readable name.

• comment (Optional[str]) –

  Field description.

• nullable (bool) –

  If this field can be NULL or not.

• length (Optional[int]) –

  The limitation in length on the field value.

• precision (Optional[int]) –

  The precision of the field value (float types)

• editor_widget_type (Optional[str]) –

  The original type how it is defined in the QGIS form.

• editor_widget_type_wfs (Optional[str]) –

  The translated type based on the simple types of XSD spec.

• editor_widget_type_oapif (Optional[str]) –

  Translated type based on the types of the OpenAPI Spec

• editor_widget_type_oapif_format (Optional[str]) –

  Format of the above-mentioned type based on the OpenAPI Spec

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Field(BaseInterface):
    """
    Transportable (serializable) form of a QGIS vector job_layer_definition fiel (attribute). It contains the information of
    the original data datatype and its translated versions and the editor widget one as well.

    Attributes:
        name: Machine readable name of the field
        type: Original type as defined by data source (PostGIS, GPKG, etc.)
        is_primary_key: if the field is considered to be primary key.
        type_wfs: Translated type for further usage. Based on the simple types of
            [XSD spec](https://www.w3.org/TR/xmlschema11-2/#built-in-primitive-datatypes).
        type_oapif: Translated type based on the types of the
            [OpenAPI Spec](https://spec.openapis.org/oas/latest.html#data-types)
        type_oapif_format: Format of the above-mentioned type based on the
            [OpenAPI Spec](https://spec.openapis.org/oas/latest.html#data-types)
        alias: Human readable name.
        comment: Field description.
        nullable: If this field can be NULL or not.
        length: The limitation in length on the field value.
        precision: The precision of the field value (float types)
        editor_widget_type: The original type how it is defined in the QGIS form.
        editor_widget_type_wfs: The translated type based on the simple types of
            [XSD spec](https://www.w3.org/TR/xmlschema11-2/#built-in-primitive-datatypes).
        editor_widget_type_oapif: Translated type based on the types of the
            [OpenAPI Spec](https://spec.openapis.org/oas/latest.html#data-types)
        editor_widget_type_oapif_format: Format of the above-mentioned type based on the
            [OpenAPI Spec](https://spec.openapis.org/oas/latest.html#data-types)
    """

    name: str = field(metadata={"type": "Element"})
    type: str = field(metadata={"type": "Element"})
    is_primary_key: bool = field(
        default=False,
        metadata={"type": "Element"},
    )
    type_wfs: Optional[str] = field(default=None, metadata={"type": "Element"})
    type_oapif: Optional[str] = field(default=None, metadata={"type": "Element"})
    type_oapif_format: Optional[str] = field(default=None, metadata={"type": "Element"})
    alias: Optional[str] = field(default=None, metadata={"type": "Element"})
    comment: Optional[str] = field(default=None, metadata={"type": "Element"})
    nullable: bool = field(default=True, metadata={"type": "Element"})
    length: Optional[int] = field(default=None, metadata={"type": "Element"})
    precision: Optional[int] = field(default=None, metadata={"type": "Element"})
    editor_widget_type: Optional[str] = field(
        default=None, metadata={"type": "Element"}
    )
    editor_widget_type_wfs: Optional[str] = field(
        default=None, metadata={"type": "Element"}
    )
    editor_widget_type_oapif: Optional[str] = field(
        default=None, metadata={"type": "Element"}
    )
    editor_widget_type_oapif_format: Optional[str] = field(
        default=None, metadata={"type": "Element"}
    )

alias: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

comment: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

editor_widget_type: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

editor_widget_type_oapif: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

editor_widget_type_oapif_format: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

editor_widget_type_wfs: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

is_primary_key: bool = field(default=False, metadata={'type': 'Element'}) class-attribute instance-attribute

length: Optional[int] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

nullable: bool = field(default=True, metadata={'type': 'Element'}) class-attribute instance-attribute

precision: Optional[int] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

type: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

type_oapif: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

type_oapif_format: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

type_wfs: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(name: str, type: str, is_primary_key: bool = False, type_wfs: Optional[str] = None, type_oapif: Optional[str] = None, type_oapif_format: Optional[str] = None, alias: Optional[str] = None, comment: Optional[str] = None, nullable: bool = True, length: Optional[int] = None, precision: Optional[int] = None, editor_widget_type: Optional[str] = None, editor_widget_type_wfs: Optional[str] = None, editor_widget_type_oapif: Optional[str] = None, editor_widget_type_oapif_format: Optional[str] = None) -> None

GdalSource dataclass

Bases: Source

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class GdalSource(Source):
    path: str = field(metadata={"type": "Element"})
    layer_name: str | None = field(default=None, metadata={"type": "Element"})
    vsi_prefix: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def remote(self):
        return self.decide_remote(self.path)

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = {"path": self.path}
        if self.layer_name is not None:
            connection_dict["layerName"] = self.layer_name
        if self.vsi_prefix is not None:
            connection_dict["vsiPrefix"] = self.vsi_prefix
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        return cls(
            path=decoded_uri["path"],
            layer_name=decoded_uri.get("layerName"),
            vsi_prefix=decoded_uri.get("vsiPrefix"),
        )

layer_name: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

path: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

remote property

to_qgis_decoded_uri: dict property

vsi_prefix: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(path: str, layer_name: str | None = None, vsi_prefix: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    return cls(
        path=decoded_uri["path"],
        layer_name=decoded_uri.get("layerName"),
        vsi_prefix=decoded_uri.get("vsiPrefix"),
    )

Group dataclass

Bases: AbstractDataset

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Group(AbstractDataset):
    pass

__init__(id: str, name: str, title: str) -> None

LayerLike dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class LayerLike(BaseInterface):
    id: str = field(metadata={"type": "Element"})
    name: str = field(metadata={"type": "Element"})

id: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, name: str) -> None

MetaData dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class MetaData(BaseInterface):
    service: Service = field(metadata={"type": "Element"})
    links: Optional[List[str]] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )
    language: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    categories: Optional[List[str]] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )
    creationDateTime: str = field(
        default=None,
        metadata={"type": "Element"},
    )
    author: Optional[str] = field(default=None, metadata={"type": "Element"})

    def __post_init__(self):
        if self.creationDateTime is None:
            self.creationDateTime = datetime.now(UTC).isoformat()

author: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

categories: Optional[List[str]] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

creationDateTime: str = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

language: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

links: Optional[List[str]] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

service: Service = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(service: Service, links: Optional[List[str]] = list(), language: Optional[str] = None, categories: Optional[List[str]] = list(), creationDateTime: str = None, author: Optional[str] = None) -> None

__post_init__()

Source code in src/qgis_server_light/interface/exporter/extract.py

def __post_init__(self):
    if self.creationDateTime is None:
        self.creationDateTime = datetime.now(UTC).isoformat()

OgrSource dataclass

Bases: GdalSource

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class OgrSource(GdalSource):
    layer_id: str | None = field(default=None, metadata={"type": "Element"})
    subset: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = super().to_qgis_decoded_uri
        if self.layer_id:
            connection_dict["layerId"] = self.layer_id
        if self.subset:
            connection_dict["subset"] = self.subset
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        base_class_instance = GdalSource.from_qgis_decoded_uri(decoded_uri)
        return cls(
            path=base_class_instance.path,
            layer_name=base_class_instance.layer_name,
            vsi_prefix=base_class_instance.vsi_prefix,
            layer_id=decoded_uri.get("layerId"),
            subset=decoded_uri.get("subset"),
        )

    @property
    def encoded_uri_separator(self) -> str:
        return "|"

encoded_uri_separator: str property

layer_id: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

subset: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

to_qgis_decoded_uri: dict property

__init__(path: str, layer_name: str | None = None, vsi_prefix: str | None = None, layer_id: str | None = None, subset: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    base_class_instance = GdalSource.from_qgis_decoded_uri(decoded_uri)
    return cls(
        path=base_class_instance.path,
        layer_name=base_class_instance.layer_name,
        vsi_prefix=base_class_instance.vsi_prefix,
        layer_id=decoded_uri.get("layerId"),
        subset=decoded_uri.get("subset"),
    )

PostgresSource dataclass

Bases: Source

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class PostgresSource(Source):
    key: str = field(metadata={"type": "Element"})
    schema: str = field(metadata={"type": "Element"})
    table: str = field(metadata={"type": "Element"})
    geometry_column: str | None = field(default=None, metadata={"type": "Element"})
    dbname: str | None = field(default=None, metadata={"type": "Element"})
    host: str | None = field(default=None, metadata={"type": "Element"})
    password: str | None = field(default=None, metadata={"type": "Element"})
    port: int | None = field(default=None, metadata={"type": "Element"})
    type: int | None = field(default=None, metadata={"type": "Element"})
    username: str | None = field(default=None, metadata={"type": "Element"})
    srid: str | None = field(default=None, metadata={"type": "Element"})
    sslmode: int | None = field(default=None, metadata={"type": "Element"})
    ssl_mode_text: str | None = field(default=None, metadata={"type": "Element"})
    service: str | None = field(default=None, metadata={"type": "Element"})
    check_primary_key_unicity: str | None = field(
        default=None, metadata={"type": "Element"}
    )
    sql: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def redacted_fields(self) -> set:
        return {"password"}

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = {"key": self.key, "schema": self.schema, "table": self.table}
        if self.geometry_column is not None:
            connection_dict["geometrycolumn"] = self.geometry_column
        if self.dbname is not None:
            connection_dict["dbname"] = self.dbname
        if self.host is not None:
            connection_dict["host"] = self.host
        if self.password is not None:
            connection_dict["password"] = self.password
        if self.port is not None:
            connection_dict["port"] = self.port
        if self.type is not None:
            connection_dict["type"] = self.type
        if self.username is not None:
            connection_dict["username"] = self.username
        if self.srid is not None:
            connection_dict["srid"] = self.srid
        if self.sslmode is not None:
            connection_dict["sslmode"] = self.sslmode
        if self.service is not None:
            connection_dict["service"] = self.service
        if self.check_primary_key_unicity is not None:
            connection_dict["checkPrimaryKeyUnicity"] = self.check_primary_key_unicity
        if self.sql is not None:
            connection_dict["sql"] = self.sql
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        return cls(
            key=decoded_uri["key"],
            schema=decoded_uri["schema"],
            table=decoded_uri["table"],
            geometry_column=decoded_uri.get("geometrycolumn"),
            dbname=decoded_uri.get("dbname"),
            host=decoded_uri.get("host"),
            password=decoded_uri.get("password"),
            port=int(decoded_uri.get("port"))
            if decoded_uri.get("port") is not None
            else None,
            type=int(decoded_uri.get("type"))
            if decoded_uri.get("type") is not None
            else None,
            username=decoded_uri.get("username"),
            srid=decoded_uri.get("srid"),
            sslmode=int(decoded_uri.get("sslmode", 2)),
            service=decoded_uri.get("service"),
            check_primary_key_unicity=decoded_uri.get("check_primary_key_unicity"),
            sql=decoded_uri.get("sql"),
        )

    @property
    def remote(self):
        return True

check_primary_key_unicity: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

dbname: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

geometry_column: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

host: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

key: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

password: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

port: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

redacted_fields: set property

remote property

schema: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

service: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

sql: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

srid: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

ssl_mode_text: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

sslmode: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

table: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

to_qgis_decoded_uri: dict property

type: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

username: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(key: str, schema: str, table: str, geometry_column: str | None = None, dbname: str | None = None, host: str | None = None, password: str | None = None, port: int | None = None, type: int | None = None, username: str | None = None, srid: str | None = None, sslmode: int | None = None, ssl_mode_text: str | None = None, service: str | None = None, check_primary_key_unicity: str | None = None, sql: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    return cls(
        key=decoded_uri["key"],
        schema=decoded_uri["schema"],
        table=decoded_uri["table"],
        geometry_column=decoded_uri.get("geometrycolumn"),
        dbname=decoded_uri.get("dbname"),
        host=decoded_uri.get("host"),
        password=decoded_uri.get("password"),
        port=int(decoded_uri.get("port"))
        if decoded_uri.get("port") is not None
        else None,
        type=int(decoded_uri.get("type"))
        if decoded_uri.get("type") is not None
        else None,
        username=decoded_uri.get("username"),
        srid=decoded_uri.get("srid"),
        sslmode=int(decoded_uri.get("sslmode", 2)),
        service=decoded_uri.get("service"),
        check_primary_key_unicity=decoded_uri.get("check_primary_key_unicity"),
        sql=decoded_uri.get("sql"),
    )

Project dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Project(BaseInterface):
    version: str = field(metadata={"type": "Element"})
    name: str = field(metadata={"type": "Element"})

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

version: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(version: str, name: str) -> None

Raster dataclass

Bases: DataSet

A real QGIS Raster job_layer_definition. That are usually all QgsRasterLayer in opposition to QgsVectorTileLayer which is not a real QgsRasterLayer.

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Raster(DataSet):
    """
    A real QGIS Raster job_layer_definition. That are usually all `QgsRasterLayer` in opposition to `QgsVectorTileLayer`
    which is not a real `QgsRasterLayer`.
    """

__init__(id: str, name: str, title: str, source: DataSource, driver: str, bbox: BBox | None = None, bbox_wgs84: BBox | None = None, crs: Crs | None = None, styles: List[Style] = list(), minimum_scale: float | None = None, maximum_scale: float | None = None, style_name: str = 'default', is_spatial: bool = True) -> None

Service dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Service(BaseInterface):
    contact_organization: Optional[str] = field(metadata={"type": "Element"})
    contact_mail: Optional[str] = field(metadata={"type": "Element"})
    contact_person: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    contact_phone: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    contact_position: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    fees: Optional[str] = field(default=None, metadata={"type": "Element"})
    keyword_list: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    online_resource: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    service_abstract: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    service_title: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    resource_url: Optional[str] = field(default=None, metadata={"type": "Element"})

contact_mail: Optional[str] = field(metadata={'type': 'Element'}) class-attribute instance-attribute

contact_organization: Optional[str] = field(metadata={'type': 'Element'}) class-attribute instance-attribute

contact_person: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

contact_phone: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

contact_position: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

fees: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

keyword_list: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

online_resource: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

resource_url: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

service_abstract: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

service_title: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(contact_organization: Optional[str], contact_mail: Optional[str], contact_person: Optional[str] = None, contact_phone: Optional[str] = None, contact_position: Optional[str] = None, fees: Optional[str] = None, keyword_list: Optional[str] = None, online_resource: Optional[str] = None, service_abstract: Optional[str] = None, service_title: Optional[str] = None, resource_url: Optional[str] = None) -> None

Source dataclass

Bases: BaseInterface, ABC

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Source(BaseInterface, ABC):
    @staticmethod
    def decide_remote(path: str) -> bool:
        return path.startswith("http")

    @property
    def to_qgis_decoded_uri(self) -> dict:
        raise NotImplementedError(
            "This is a base class, the method has to be defined at implementation level."
        )

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        raise NotImplementedError(
            "This is a base class, the method has to be defined at implementation level."
        )

to_qgis_decoded_uri: dict property

__init__() -> None

decide_remote(path: str) -> bool staticmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@staticmethod
def decide_remote(path: str) -> bool:
    return path.startswith("http")

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    raise NotImplementedError(
        "This is a base class, the method has to be defined at implementation level."
    )

Tree dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Tree(BaseInterface):
    members: list[TreeGroup] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )

    def find_by_name(self, name: str) -> TreeGroup | None:
        for member in self.members:
            if member.name == name:
                return member
        return None

members: list[TreeGroup] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(members: list[TreeGroup] = list()) -> None

find_by_name(name: str) -> TreeGroup | None

Source code in src/qgis_server_light/interface/exporter/extract.py

def find_by_name(self, name: str) -> TreeGroup | None:
    for member in self.members:
        if member.name == name:
            return member
    return None

TreeGroup dataclass

Bases: TreeLayer

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class TreeGroup(TreeLayer):
    children: List[str] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )

children: List[str] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, name: str, children: List[str] = list()) -> None

TreeLayer dataclass

Bases: LayerLike

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class TreeLayer(LayerLike):
    pass

__init__(id: str, name: str) -> None

Vector dataclass

Bases: DataSet

A real QGIS Vector job_layer_definition. That are usually all QgsVectorLayer in opposition to QgsVectorTileLayer which is not a real QgsVectorLayer.

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class Vector(DataSet):
    """
    A real QGIS Vector job_layer_definition. That are usually all `QgsVectorLayer` in opposition to `QgsVectorTileLayer`
    which is not a real `QgsVectorLayer`.
    """

    fields: Optional[List[Field]] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )
    geometry_type_simple: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )
    geometry_type_wkb: Optional[str] = field(
        default=None,
        metadata={"type": "Element"},
    )

    def get_field_by_name(self, name: str) -> Field | None:
        for dataclass_field in self.fields:
            if dataclass_field.name == name:
                return dataclass_field
        return None

fields: Optional[List[Field]] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

geometry_type_simple: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

geometry_type_wkb: Optional[str] = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, name: str, title: str, source: DataSource, driver: str, bbox: BBox | None = None, bbox_wgs84: BBox | None = None, crs: Crs | None = None, styles: List[Style] = list(), minimum_scale: float | None = None, maximum_scale: float | None = None, style_name: str = 'default', is_spatial: bool = True, fields: Optional[List[Field]] = list(), geometry_type_simple: Optional[str] = None, geometry_type_wkb: Optional[str] = None) -> None

get_field_by_name(name: str) -> Field | None

Source code in src/qgis_server_light/interface/exporter/extract.py

def get_field_by_name(self, name: str) -> Field | None:
    for dataclass_field in self.fields:
        if dataclass_field.name == name:
            return dataclass_field
    return None

VectorTileSource dataclass

Bases: Source

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class VectorTileSource(Source):
    type: str = field(metadata={"type": "Element"})
    zmin: int | None = field(metadata={"type": "Element"})
    zmax: int | None = field(metadata={"type": "Element"})
    url: str | None = field(default=None, metadata={"type": "Element"})
    path: str | None = field(default=None, metadata={"type": "Element"})
    style_url: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def remote(self):
        return self.decide_remote(self.url)

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = {"type": self.type, "zmin": self.zmin, "zmax": self.zmax}
        if self.url is not None:
            connection_dict["url"] = self.url
        if self.path is not None:
            connection_dict["path"] = self.path
        if self.style_url is not None:
            connection_dict["styleUrl"] = self.style_url
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        return cls(
            type=decoded_uri["type"],
            zmin=decoded_uri.get("zmin"),
            zmax=decoded_uri.get("zmax"),
            url=decoded_uri.get("url"),
            path=decoded_uri.get("path"),
            style_url=decoded_uri.get("styleUrl"),
        )

path: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

remote property

style_url: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

to_qgis_decoded_uri: dict property

type: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

url: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

zmax: int | None = field(metadata={'type': 'Element'}) class-attribute instance-attribute

zmin: int | None = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(type: str, zmin: int | None, zmax: int | None, url: str | None = None, path: str | None = None, style_url: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    return cls(
        type=decoded_uri["type"],
        zmin=decoded_uri.get("zmin"),
        zmax=decoded_uri.get("zmax"),
        url=decoded_uri.get("url"),
        path=decoded_uri.get("path"),
        style_url=decoded_uri.get("styleUrl"),
    )

WfsSource dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class WfsSource(BaseInterface):
    # currently not implemented because qgis does not allow to
    # use the decode uri approach on that URI
    pass

__init__() -> None

WmsSource dataclass

Bases: Source

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class WmsSource(Source):
    crs: str = field(metadata={"type": "Element"})
    format: str = field(metadata={"type": "Element"})
    layers: str = field(metadata={"type": "Element"})
    url: str = field(metadata={"type": "Element"})
    dpi_mode: str | None = field(default=None, metadata={"type": "Element"})
    feature_count: int | None = field(default=None, metadata={"type": "Element"})
    contextual_wms_legend: str | None = field(
        default=None, metadata={"type": "Element"}
    )
    styles: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = {
            "crs": self.crs,
            "format": self.format,
            "layers": self.layers,
            "url": self.url,
            "styles": self.styles,
        }
        if self.dpi_mode is not None:
            connection_dict["dpiMode"] = self.dpi_mode
        if self.feature_count is not None:
            connection_dict["featureCount"] = self.feature_count
        if self.contextual_wms_legend is not None:
            connection_dict["contextualWMSLegend"] = self.contextual_wms_legend
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        return cls(
            crs=decoded_uri["crs"],
            format=decoded_uri["format"],
            layers=decoded_uri["layers"],
            url=decoded_uri["url"],
            dpi_mode=decoded_uri.get("dpiMode"),
            feature_count=decoded_uri.get("featureCount"),
            contextual_wms_legend=decoded_uri.get("contextualWMSLegend"),
            styles=decoded_uri.get("styles"),
        )

    @property
    def remote(self):
        return True

contextual_wms_legend: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

crs: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

dpi_mode: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

feature_count: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

format: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

layers: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

remote property

styles: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

to_qgis_decoded_uri: dict property

url: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(crs: str, format: str, layers: str, url: str, dpi_mode: str | None = None, feature_count: int | None = None, contextual_wms_legend: str | None = None, styles: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    return cls(
        crs=decoded_uri["crs"],
        format=decoded_uri["format"],
        layers=decoded_uri["layers"],
        url=decoded_uri["url"],
        dpi_mode=decoded_uri.get("dpiMode"),
        feature_count=decoded_uri.get("featureCount"),
        contextual_wms_legend=decoded_uri.get("contextualWMSLegend"),
        styles=decoded_uri.get("styles"),
    )

WmtsSource dataclass

Bases: WmsSource

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False, kw_only=True)
class WmtsSource(WmsSource):
    tile_matrix_set: str = field(metadata={"type": "Element"})
    tile_dimensions: str | None = field(
        default=None,
        metadata={"type": "Element"},
    )
    tile_pixel_ratio: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = super().to_qgis_decoded_uri
        connection_dict["tileMatrixSet"] = self.tile_matrix_set
        if self.tile_dimensions is not None:
            connection_dict["tileDimensions"] = self.tile_dimensions
        if self.tile_pixel_ratio is not None:
            connection_dict["tilePixelRatio"] = self.tile_pixel_ratio
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        base_class_instance = WmsSource.from_qgis_decoded_uri(decoded_uri)
        return cls(
            crs=base_class_instance.crs,
            format=base_class_instance.format,
            layers=base_class_instance.layers,
            url=base_class_instance.url,
            dpi_mode=base_class_instance.dpi_mode,
            feature_count=base_class_instance.feature_count,
            contextual_wms_legend=base_class_instance.contextual_wms_legend,
            styles=base_class_instance.styles,
            tile_matrix_set=decoded_uri["tileMatrixSet"],
            tile_dimensions=decoded_uri.get("tileDimensions"),
            tile_pixel_ratio=decoded_uri.get("tilePixelRatio"),
        )

tile_dimensions: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

tile_matrix_set: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

tile_pixel_ratio: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

to_qgis_decoded_uri: dict property

__init__(crs: str, format: str, layers: str, url: str, dpi_mode: str | None = None, feature_count: int | None = None, contextual_wms_legend: str | None = None, styles: str | None = None, *, tile_matrix_set: str, tile_dimensions: str | None = None, tile_pixel_ratio: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    base_class_instance = WmsSource.from_qgis_decoded_uri(decoded_uri)
    return cls(
        crs=base_class_instance.crs,
        format=base_class_instance.format,
        layers=base_class_instance.layers,
        url=base_class_instance.url,
        dpi_mode=base_class_instance.dpi_mode,
        feature_count=base_class_instance.feature_count,
        contextual_wms_legend=base_class_instance.contextual_wms_legend,
        styles=base_class_instance.styles,
        tile_matrix_set=decoded_uri["tileMatrixSet"],
        tile_dimensions=decoded_uri.get("tileDimensions"),
        tile_pixel_ratio=decoded_uri.get("tilePixelRatio"),
    )

XYZSource dataclass

Bases: Source

Source code in src/qgis_server_light/interface/exporter/extract.py

@dataclass(repr=False)
class XYZSource(Source):
    url: str = field(metadata={"type": "Element"})
    zmin: int | None = field(default=None, metadata={"type": "Element"})
    zmax: int | None = field(default=None, metadata={"type": "Element"})
    type: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def to_qgis_decoded_uri(self) -> dict:
        connection_dict = {
            "url": self.url,
            "zmin": self.zmin,
            "zmax": self.zmax,
            "type": self.type,
        }
        return connection_dict

    @classmethod
    def from_qgis_decoded_uri(cls, decoded_uri: dict):
        return cls(
            url=decoded_uri["url"],
            zmin=decoded_uri.get("zmin"),
            zmax=decoded_uri.get("zmax"),
            type=decoded_uri.get("type"),
        )

    @property
    def remote(self):
        return self.decide_remote(self.url)

remote property

to_qgis_decoded_uri: dict property

type: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

url: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

zmax: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

zmin: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(url: str, zmin: int | None = None, zmax: int | None = None, type: str | None = None) -> None

from_qgis_decoded_uri(decoded_uri: dict) classmethod

Source code in src/qgis_server_light/interface/exporter/extract.py

@classmethod
def from_qgis_decoded_uri(cls, decoded_uri: dict):
    return cls(
        url=decoded_uri["url"],
        zmin=decoded_uri.get("zmin"),
        zmax=decoded_uri.get("zmax"),
        type=decoded_uri.get("type"),
    )

job

common

input

AbstractFilter dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass(repr=False)
class AbstractFilter(BaseInterface):
    definition: str = field(metadata={"type": "Element"})

    @property
    def shortened_fields(self) -> set:
        return {"definition"}

definition: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

shortened_fields: set property

__init__(definition: str) -> None

OgcFilter110 dataclass

Bases: AbstractFilter

A filter which definition conforms to https://schemas.opengis.net/filter/1.1.0/filter.xsd and which is consumable by qgis.core.QgsOgcUtils.expressionFromOgcFilter.

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass(repr=False)
class OgcFilter110(AbstractFilter):
    """
    A filter which definition conforms to
    https://schemas.opengis.net/filter/1.1.0/filter.xsd
    and which is consumable by `qgis.core.QgsOgcUtils.expressionFromOgcFilter`.
    """

__init__(definition: str) -> None

OgcFilterFES20 dataclass

Bases: AbstractFilter

A filter which definition conforms to https://www.opengis.net/fes/2.0 and which is consumable by qgis.core.QgsOgcUtils.expressionFromOgcFilter.

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass(repr=False)
class OgcFilterFES20(AbstractFilter):
    """
    A filter which definition conforms to https://www.opengis.net/fes/2.0
    and which is consumable by `qgis.core.QgsOgcUtils.expressionFromOgcFilter`.
    """

__init__(definition: str) -> None

QslJobInfoParameter dataclass

Bases: ABC

The common minimal interface of a job which is shipped around. Each job for QSL has to implement at least this interface.

Attributes:

• id (str) –

  The unique identifier which is used to recognize the job all over its lifecycle.

• type (str) –

  A string based identifier of the job, this is used to quickly determine its nature serialized state.

• job (QslJobParameter) –

  The actual job parameters. This is a domain specific dataclass depending on the nature of the actual job.

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass
class QslJobInfoParameter(ABC):
    """The common minimal interface of a job which is
    shipped around. Each job for QSL has to implement at least this
    interface.

    Attributes:
        id: The unique identifier which is used to recognize the job
            all over its lifecycle.
        type: A string based identifier of the job, this is used to quickly
            determine its nature serialized state.
        job: The actual job parameters. This is a domain specific dataclass
            depending on the nature of the actual job.
    """

    id: str = field(metadata={"type": "Element"})
    type: str = field(metadata={"type": "Element"})
    job: QslJobParameter = field(metadata={"type": "Element"})

id: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

job: QslJobParameter = field(metadata={'type': 'Element'}) class-attribute instance-attribute

type: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, type: str, job: QslJobParameter) -> None

QslJobLayer dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass(repr=False)
class QslJobLayer(BaseInterface):
    id: str = field(metadata={"type": "Element"})
    name: str = field(metadata={"type": "Element"})
    source: str = field(metadata={"type": "Element"})
    remote: bool = field(metadata={"type": "Element"})
    folder_name: str = field(metadata={"type": "Element"})
    driver: str = field(metadata={"type": "Element"})
    style: Style | None = field(default=None, metadata={"type": "Element"})
    filter: OgcFilter110 | OgcFilterFES20 | None = field(
        default=None, metadata={"type": "Element"}
    )

    @property
    def redacted_fields(self) -> set:
        return {"source"}

driver: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

filter: OgcFilter110 | OgcFilterFES20 | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

folder_name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

id: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

redacted_fields: set property

remote: bool = field(metadata={'type': 'Element'}) class-attribute instance-attribute

source: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

style: Style | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, name: str, source: str, remote: bool, folder_name: str, driver: str, style: Style | None = None, filter: OgcFilter110 | OgcFilterFES20 | None = None) -> None

QslJobParameter dataclass

Bases: ABC

The minimal interface of a job parameter interface. In the domain specific refinement it holds the relevant information about a job.

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass
class QslJobParameter(ABC):
    """The minimal interface of a job parameter interface. In the domain
    specific refinement it holds the relevant information about a job.
    """

    pass

__init__() -> None

QslJobParameterMapRelated dataclass

Bases: QslJobParameter

The minimal interface of a job parameter interface for jobs rendering things in the end.

Attributes:

• svg_paths (list[str]) –

  A list of paths to svg's (folders) which are necessary for the job to render nicely.

Source code in src/qgis_server_light/interface/job/common/input.py

@dataclass
class QslJobParameterMapRelated(QslJobParameter):
    """The minimal interface of a job parameter interface for jobs rendering things in the end.

    Attributes:
        svg_paths: A list of paths to svg's (folders) which are necessary for
            the job to render nicely.
    """

    svg_paths: list[str] = field(default_factory=list, metadata={"type": "Element"})

svg_paths: list[str] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(svg_paths: list[str] = list()) -> None

output

JobResult dataclass

Bases: BaseInterface

Source code in src/qgis_server_light/interface/job/common/output.py

@dataclass
class JobResult(BaseInterface):
    id: str = field(metadata={"type": "Element"})
    data: Any = field(metadata={"type": "Element"})
    content_type: str = field(metadata={"type": "Element"})
    worker_id: str | None = field(default=None, metadata={"type": "Element"})
    worker_host_name: str | None = field(default=None, metadata={"type": "Element"})

    @property
    def shortened_fields(self) -> set:
        return {"data"}

content_type: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

data: Any = field(metadata={'type': 'Element'}) class-attribute instance-attribute

id: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

shortened_fields: set property

worker_host_name: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

worker_id: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, data: Any, content_type: str, worker_id: str | None = None, worker_host_name: str | None = None) -> None

feature

input

FeatureQuery dataclass

Bases: BaseInterface

Represents definitions of a query to obtain features from a list of layers. Be aware, that filters are not applied to the QslJobLayer in this implementation since passed filters can contain inter-layer-references.

Attributes:

• layers (list[QslJobLayer]) –

  A list layers which should only reference vector sources and be queried.

• aliases (list[str]) –

  An optional list of alias names. This has to be the same length as the list of datasets.

• filter (OgcFilterFES20) –

  An optional filter which might reference all passed layers thats why layers has to be added

Source code in src/qgis_server_light/interface/job/feature/input.py

@dataclass
class FeatureQuery(BaseInterface):
    """Represents definitions of a query to obtain features from a list of layers.
    Be aware, that filters are not applied to the QslJobLayer in this implementation
    since passed filters can contain inter-layer-references.

    Attributes:
        layers: A list layers which should only reference vector sources and be queried.
        aliases: An optional list of alias names. This has to be the same length as the list of datasets.
        filter: An optional filter which might reference all passed layers thats why layers
            has to be added
    """

    layers: list[QslJobLayer] = field(metadata={"type": "Element"})
    aliases: list[str] = field(default_factory=list, metadata={"type": "Element"})
    filter: OgcFilterFES20 = field(default=None, metadata={"type": "Element"})

aliases: list[str] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

filter: OgcFilterFES20 = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

layers: list[QslJobLayer] = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(layers: list[QslJobLayer], aliases: list[str] = list(), filter: OgcFilterFES20 = None) -> None

QslJobInfoFeature dataclass

Bases: QslJobInfoParameter

Source code in src/qgis_server_light/interface/job/feature/input.py

@dataclass
class QslJobInfoFeature(QslJobInfoParameter):
    job: QslJobParameterFeature = field(metadata={"type": "Element"})

job: QslJobParameterFeature = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, type: str, job: QslJobParameterFeature) -> None

QslJobParameterFeature dataclass

Bases: QslJobParameter

As defined in WFS 2.0 specs, a request can be subdivided in a list of queries. This class is representing that.

Attributes:

• queries (list[FeatureQuery]) –

  A list of queries which features should be extracted for.

• start_index (int) –

  The offset for paging reason.

• count (int | None) –

  The number of results to return.

Source code in src/qgis_server_light/interface/job/feature/input.py

@dataclass
class QslJobParameterFeature(QslJobParameter):
    """As defined in WFS 2.0 specs, a request can be subdivided in a list of queries.
    This class is representing that.

    Attributes:
        queries: A list of queries which features should be extracted for.
        start_index: The offset for paging reason.
        count: The number of results to return.
    """

    queries: list[FeatureQuery] = field(metadata={"type": "Element"})
    start_index: int = field(
        default=0,
        metadata={
            "type": "Element",
        },
    )
    count: int | None = field(
        default=None,
        metadata={
            "type": "Element",
        },
    )

count: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

queries: list[FeatureQuery] = field(metadata={'type': 'Element'}) class-attribute instance-attribute

start_index: int = field(default=0, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(queries: list[FeatureQuery], start_index: int = 0, count: int | None = None) -> None

output

Attribute dataclass

Bases: BaseInterface

An attribute belonging to a feature. The aim here is to drill down to simple types which can be used in consuming applications without further handling. This does not include the geometry attribute!

Attributes:

• name (str) –

  The name of the attribute. Has to match with the name used for exported fields with Field class.

• value (int | float | str | bool | bytes | None) –

  Value as simple as possible. It has to be pickleable

Source code in src/qgis_server_light/interface/job/feature/output.py

@dataclass(repr=False)
class Attribute(BaseInterface):
    """An attribute belonging to a feature. The aim here is to
    drill down to simple types which can be used in consuming
    applications without further handling. This does not include
    the geometry attribute!

    Attributes:
        name: The name of the attribute. Has to match with the
            name used for exported fields with `Field` class.
        value: Value as simple as possible. It has to be
            [pickleable](https://docs.python.org/3/library/pickle.html#what-can-be-pickled-and-unpickled)

    """

    name: str = field(metadata={"type": "Element"})
    value: int | float | str | bool | bytes | None = field(
        metadata={"type": "Element", "format": "base64"}
    )

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

value: int | float | str | bool | bytes | None = field(metadata={'type': 'Element', 'format': 'base64'}) class-attribute instance-attribute

__init__(name: str, value: int | float | str | bool | bytes | None) -> None

Feature dataclass

Bases: BaseInterface

Feature to hold information of extracted QgsFeature.

Attributes:

• geometry (Geometry | None) –

  The geometry representing the feature.

• attributes (list[Attribute]) –

  List of attributes defined in this feature.

Source code in src/qgis_server_light/interface/job/feature/output.py

@dataclass(repr=False)
class Feature(BaseInterface):
    """Feature to hold information of extracted QgsFeature.

    Attributes:
        geometry: The geometry representing the feature.
        attributes: List of attributes defined in this feature.
    """

    geometry: Geometry | None = field(default=None, metadata={"type": "Element"})
    attributes: list[Attribute] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )

attributes: list[Attribute] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

geometry: Geometry | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(geometry: Geometry | None = None, attributes: list[Attribute] = list()) -> None

FeatureCollection dataclass

Bases: BaseInterface

This construction is used to abstract the content of extracted features for pickelable transportation from QSL to the queue. This way we ensure how things are constructed and transported.

Attributes:

• name (str) –

  The name of the feature collection. This is the key to match it to requested layers.

• features (list[Feature]) –

  The features belonging to the feature collection.

Source code in src/qgis_server_light/interface/job/feature/output.py

@dataclass(repr=False)
class FeatureCollection(BaseInterface):
    """This construction is used to abstract the content of extracted
    features for pickelable transportation from QSL to the queue.
    This way we ensure how things are constructed and transported.

    Attributes:
        name: The name of the feature collection. This is the key to
            match it to requested layers.
        features: The features belonging to the feature collection.
    """

    name: str = field(metadata={"type": "Element"})
    features: list[Feature] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )

features: list[Feature] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(name: str, features: list[Feature] = list()) -> None

Geometry dataclass

Bases: Attribute

Source code in src/qgis_server_light/interface/job/feature/output.py

@dataclass(repr=False)
class Geometry(Attribute):
    name: str = field(default="geometry", metadata={"type": "Element"})
    value: bytes | None = field(
        default=None, metadata={"type": "Element", "format": "base64"}
    )

    @property
    def shortened_fields(self) -> set:
        return {"value"}

name: str = field(default='geometry', metadata={'type': 'Element'}) class-attribute instance-attribute

shortened_fields: set property

value: bytes | None = field(default=None, metadata={'type': 'Element', 'format': 'base64'}) class-attribute instance-attribute

__init__(name: str = 'geometry', value: bytes | None = None) -> None

QueryCollection dataclass

Bases: BaseInterface

Holds all feature collections which are bound to the passed queries. The order in the list has to be not changed, so that consuming applications can map the response to the passed queries.

Attributes:

• numbers_matched (str | int) –

  Information about how many matches are fund for the executed query.

• feature_collections (list[FeatureCollection]) –

  The feature collections belonging to the passed queries.

Source code in src/qgis_server_light/interface/job/feature/output.py

@dataclass(repr=False)
class QueryCollection(BaseInterface):
    """Holds all feature collections which are bound to the
    passed queries. The order in the list has to be not changed,
    so that consuming applications can map the response to the
    passed queries.

    Attributes:
        numbers_matched: Information about how many matches are fund for the executed query.
        feature_collections: The feature collections belonging to the passed queries.
    """

    numbers_matched: str | int = field(
        default="unknown",
        metadata={"type": "Element"},
    )
    feature_collections: list[FeatureCollection] = field(
        default_factory=list,
        metadata={"type": "Element"},
    )

feature_collections: list[FeatureCollection] = field(default_factory=list, metadata={'type': 'Element'}) class-attribute instance-attribute

numbers_matched: str | int = field(default='unknown', metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(numbers_matched: str | int = 'unknown', feature_collections: list[FeatureCollection] = list()) -> None

feature_info

input

QslJobInfoFeatureInfo dataclass

Bases: QslJobInfoParameter

Source code in src/qgis_server_light/interface/job/feature_info/input.py

@dataclass
class QslJobInfoFeatureInfo(QslJobInfoParameter):
    job: QslJobParameterFeatureInfo = field(
        metadata={"type": "Element", "required": True}
    )

job: QslJobParameterFeatureInfo = field(metadata={'type': 'Element', 'required': True}) class-attribute instance-attribute

__init__(id: str, type: str, job: QslJobParameterFeatureInfo) -> None

QslJobParameterFeatureInfo dataclass

Bases: QslJobParameter

A runner to extract feature info

Source code in src/qgis_server_light/interface/job/feature_info/input.py

@dataclass(kw_only=True)
class QslJobParameterFeatureInfo(QslJobParameter):
    """A runner to extract feature info"""

    # mime type, only application/json supported
    INFO_FORMAT: str = field(metadata={"type": "Element"})
    QUERY_LAYERS: str = field(metadata={"type": "Element"})
    X: str | None = field(default=None, metadata={"type": "Element"})
    Y: str | None = field(default=None, metadata={"type": "Element"})
    I: str | None = field(default=None, metadata={"type": "Element"})  # noqa: E741
    J: str | None = field(default=None, metadata={"type": "Element"})

    def __post_init__(self):
        x = int(self.I or self.X)
        y = int(self.J or self.Y)
        if x is None or y is None:
            raise KeyError(
                "Parameter `I` or `X` and `J` or `Y`  are mandatory for GetFeatureInfo"
            )
        if self.QUERY_LAYERS is None:
            raise KeyError("QUERY_LAYERS is mandatory in this request")

    @property
    def decide_x(self) -> int:
        return int(self.I or self.X)

    @property
    def decide_y(self) -> int:
        return int(self.J or self.Y)

    @property
    def query_layers_list(self):
        return self.QUERY_LAYERS.split(",")

I: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

INFO_FORMAT: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

J: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

QUERY_LAYERS: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

X: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

Y: str | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

decide_x: int property

decide_y: int property

query_layers_list property

__init__(*, INFO_FORMAT: str, QUERY_LAYERS: str, X: str | None = None, Y: str | None = None, I: str | None = None, J: str | None = None) -> None

__post_init__()

Source code in src/qgis_server_light/interface/job/feature_info/input.py

def __post_init__(self):
    x = int(self.I or self.X)
    y = int(self.J or self.Y)
    if x is None or y is None:
        raise KeyError(
            "Parameter `I` or `X` and `J` or `Y`  are mandatory for GetFeatureInfo"
        )
    if self.QUERY_LAYERS is None:
        raise KeyError("QUERY_LAYERS is mandatory in this request")

output

legend

input

QslJobInfoLegend dataclass

Bases: QslJobInfoParameter

Source code in src/qgis_server_light/interface/job/legend/input.py

@dataclass
class QslJobInfoLegend(QslJobInfoParameter):
    job: QslJobParameterLegend = field(metadata={"type": "Element", "required": True})

job: QslJobParameterLegend = field(metadata={'type': 'Element', 'required': True}) class-attribute instance-attribute

__init__(id: str, type: str, job: QslJobParameterLegend) -> None

QslJobParameterLegend dataclass

Bases: QslJobParameter

Render legend

Source code in src/qgis_server_light/interface/job/legend/input.py

@dataclass(kw_only=True)
class QslJobParameterLegend(QslJobParameter):
    """Render legend"""

__init__() -> None

output

process

input

output

render

input

QslJobInfoRender dataclass

Bases: QslJobInfoParameter

Source code in src/qgis_server_light/interface/job/render/input.py

@dataclass
class QslJobInfoRender(QslJobInfoParameter):
    job: QslJobParameterRender = field(metadata={"type": "Element", "required": True})

job: QslJobParameterRender = field(metadata={'type': 'Element', 'required': True}) class-attribute instance-attribute

__init__(id: str, type: str, job: QslJobParameterRender) -> None

QslJobParameterRender dataclass

Bases: QslJobParameter

A runner to be rendered as an image

Source code in src/qgis_server_light/interface/job/render/input.py

@dataclass(kw_only=True)
class QslJobParameterRender(QslJobParameter):
    """A runner to be rendered as an image"""

    layers: list[QslJobLayer] = field(metadata={"type": "Element"})
    bbox: BBox = field(metadata={"type": "Element"})
    crs: str = field(metadata={"type": "Element"})
    width: int = field(metadata={"type": "Element"})
    height: int = field(metadata={"type": "Element"})
    dpi: int | None = field(default=None, metadata={"type": "Element"})
    format: str = field(default="image/png", metadata={"type": "Element"})

    def get_layer_by_name(self, name: str) -> QslJobLayer:
        for layer in self.layers:
            if layer.name == name:
                return layer
        raise LookupError(f'No layer with name "{name} was found."')

bbox: BBox = field(metadata={'type': 'Element'}) class-attribute instance-attribute

crs: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

dpi: int | None = field(default=None, metadata={'type': 'Element'}) class-attribute instance-attribute

format: str = field(default='image/png', metadata={'type': 'Element'}) class-attribute instance-attribute

height: int = field(metadata={'type': 'Element'}) class-attribute instance-attribute

layers: list[QslJobLayer] = field(metadata={'type': 'Element'}) class-attribute instance-attribute

width: int = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(*, layers: list[QslJobLayer], bbox: BBox, crs: str, width: int, height: int, dpi: int | None = None, format: str = 'image/png') -> None

get_layer_by_name(name: str) -> QslJobLayer

Source code in src/qgis_server_light/interface/job/render/input.py

def get_layer_by_name(self, name: str) -> QslJobLayer:
    for layer in self.layers:
        if layer.name == name:
            return layer
    raise LookupError(f'No layer with name "{name} was found."')

output

worker

info

This part defines the structure how a running QSL worker exposes its capabilities.

EngineInfo dataclass

Source code in src/qgis_server_light/interface/worker/info.py

@dataclass
class EngineInfo:
    id: str = field(metadata={"type": "Element"})
    qgis_info: QgisInfo = field(metadata={"type": "Element"})
    status: Status = field(metadata={"type": "Element"})
    started: float = field(metadata={"type": "Element"})

id: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

qgis_info: QgisInfo = field(metadata={'type': 'Element'}) class-attribute instance-attribute

started: float = field(metadata={'type': 'Element'}) class-attribute instance-attribute

status: Status = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(id: str, qgis_info: QgisInfo, status: Status, started: float) -> None

QgisInfo dataclass

Information container to ship minimal knowledge of the underlying QGIS.

Attributes:

• version (int) –

  The integer representation of the QGIS version e.g. 34400

• version_name (str) –

  The string representation which also includes the codename e.g. "QGIS Version 4.0.0-Norrköping"

Source code in src/qgis_server_light/interface/worker/info.py

@dataclass
class QgisInfo:
    """
    Information container to ship minimal knowledge of the underlying
    QGIS.

    Attributes:
        version: The integer representation of the QGIS version e.g. 34400
        version_name: The string representation which also includes the codename e.g.
            "QGIS Version 4.0.0-Norrköping"

    """

    version: int = field(metadata={"type": "Element"})
    version_name: str = field(metadata={"type": "Element"})
    path: str = field(metadata={"type": "Element"})

path: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

version: int = field(metadata={'type': 'Element'}) class-attribute instance-attribute

version_name: str = field(metadata={'type': 'Element'}) class-attribute instance-attribute

__init__(version: int, version_name: str, path: str) -> None

Status

Bases: str, Enum

Source code in src/qgis_server_light/interface/worker/info.py

class Status(str, Enum):
    STARTING = "starting"
    CRASHED = "crashed"
    WAITING = "waiting"
    PROCESSING = "processing"

CRASHED = 'crashed' class-attribute instance-attribute

PROCESSING = 'processing' class-attribute instance-attribute

STARTING = 'starting' class-attribute instance-attribute

WAITING = 'waiting' class-attribute instance-attribute

mappers

factory

Factory

Bases: ABC

Source code in src/qgis_server_light/mappers/factory.py

class Factory(ABC):
    def from_dataclass(self):
        raise NotImplementedError()

    def to_dataclass(self):
        raise NotImplementedError()

from_dataclass()

Source code in src/qgis_server_light/mappers/factory.py

def from_dataclass(self):
    raise NotImplementedError()

to_dataclass()

Source code in src/qgis_server_light/mappers/factory.py

def to_dataclass(self):
    raise NotImplementedError()

worker

engine

Engine

Bases: ABC

Source code in src/qgis_server_light/worker/engine.py

class Engine(ABC):
    def __init__(
        self,
        context: EngineContext,
        runner_plugins: list[str],
        svg_paths: Optional[List[str]] = None,
        log_level=logging.WARNING,
    ):
        self.qgis = Qgis(svg_paths, log_level)
        self.context = context
        self.layer_cache: dict[Any, Any] = {}
        self.available_runner_classes: dict[str, Type[Runner]] = {}
        self.available_runner_classes_by_job_info: dict[str, Type[Runner]] = {}
        self.available_job_info_classes: dict[str, Type[QslJobInfoParameter]] = {}
        self._load_runner_plugins(runner_plugins)
        logging.debug(self.available_runner_classes)
        logging.debug(self.available_runner_classes_by_job_info)
        logging.debug(self.available_job_info_classes)
        self.info = self._initialize_infos()

    def __del__(self):
        self.qgis.exitQgis()

    def _load_runner_plugins(self, worker_plugins: list[str]):
        for path in worker_plugins:
            loaded_class = self._load_runner_class(path)
            if loaded_class is not None:
                self.available_runner_classes[path] = loaded_class
                self.available_runner_classes_by_job_info[
                    loaded_class.job_info_class.__name__
                ] = loaded_class
                self.available_job_info_classes[
                    loaded_class.job_info_class.__name__
                ] = loaded_class.job_info_class

    @staticmethod
    def _load_runner_class(path: str) -> Type[Runner] | None:
        """
        Loads a class dynamically at runtime, like:
        "mypackage.mymodule.MyClass"
        """

        module_path, class_name = path.rsplit(".", 1)
        module = importlib.import_module(module_path)
        cls = getattr(module, class_name, None)

        # Ensure the class was loaded correctly
        if cls is None:
            raise ImportError(
                f"Class '{class_name}' not found in module '{module_path}'."
            )
        if not inspect.isclass(cls):
            raise TypeError(f"Passed '{class_name}' is not a class.")

        if not issubclass(cls, Runner):
            raise TypeError(
                f"{cls.__name__} is not a plugin as expected (each plugin has to inherit from qgis_server_light.worker.job.common.Job)."
            )

        return cls

    def _initialize_infos(self):
        worker_info = EngineInfo(
            id=str(uuid.uuid4()),
            qgis_info=QgisInfo(
                version=version(),
                version_name=version_name(),
                path=self.qgis.prefixPath(),
            ),
            status=Status.STARTING,
            started=datetime.datetime.now().timestamp(),
        )
        logging.debug(json.dumps(asdict(worker_info), indent=2))
        return worker_info

    def runner_plugin_by_job_info(self, job_info: QslJobInfoParameter) -> Type[Runner]:
        """
        Here we decide which plugin we load dynamically out of the available ones.

        Args:
            job_info: Is the parameter instance we check the available worker classes and there the
                job_info_class at each.

        Returns:
            The selected runner class
        """
        try:
            return self.available_runner_classes_by_job_info[
                job_info.__class__.__name__
            ]
        except KeyError:
            raise RuntimeError(f"Type {type(job_info)} not supported")

    def process(self, job_info: QslJobInfoParameter) -> JobResult:
        runner_class = self.runner_plugin_by_job_info(job_info)
        runner = runner_class(
            self.qgis,
            JobContext(self.context.base_path),
            job_info,
            layer_cache=self.layer_cache,
        )
        return runner.run()

    @property
    def status(self):
        return self.info.status.value

    def set_waiting(self):
        self.info.status = Status.WAITING

    def set_crashed(self):
        self.info.status = Status.CRASHED

    def set_processing(self):
        self.info.status = Status.PROCESSING

available_job_info_classes: dict[str, Type[QslJobInfoParameter]] = {} instance-attribute

available_runner_classes: dict[str, Type[Runner]] = {} instance-attribute

available_runner_classes_by_job_info: dict[str, Type[Runner]] = {} instance-attribute

context = context instance-attribute

info = self._initialize_infos() instance-attribute

layer_cache: dict[Any, Any] = {} instance-attribute

qgis = Qgis(svg_paths, log_level) instance-attribute

status property

__del__()

Source code in src/qgis_server_light/worker/engine.py

def __del__(self):
    self.qgis.exitQgis()

__init__(context: EngineContext, runner_plugins: list[str], svg_paths: Optional[List[str]] = None, log_level=logging.WARNING)

Source code in src/qgis_server_light/worker/engine.py

def __init__(
    self,
    context: EngineContext,
    runner_plugins: list[str],
    svg_paths: Optional[List[str]] = None,
    log_level=logging.WARNING,
):
    self.qgis = Qgis(svg_paths, log_level)
    self.context = context
    self.layer_cache: dict[Any, Any] = {}
    self.available_runner_classes: dict[str, Type[Runner]] = {}
    self.available_runner_classes_by_job_info: dict[str, Type[Runner]] = {}
    self.available_job_info_classes: dict[str, Type[QslJobInfoParameter]] = {}
    self._load_runner_plugins(runner_plugins)
    logging.debug(self.available_runner_classes)
    logging.debug(self.available_runner_classes_by_job_info)
    logging.debug(self.available_job_info_classes)
    self.info = self._initialize_infos()

process(job_info: QslJobInfoParameter) -> JobResult

Source code in src/qgis_server_light/worker/engine.py

def process(self, job_info: QslJobInfoParameter) -> JobResult:
    runner_class = self.runner_plugin_by_job_info(job_info)
    runner = runner_class(
        self.qgis,
        JobContext(self.context.base_path),
        job_info,
        layer_cache=self.layer_cache,
    )
    return runner.run()

runner_plugin_by_job_info(job_info: QslJobInfoParameter) -> Type[Runner]

Here we decide which plugin we load dynamically out of the available ones.

Parameters:

• job_info (QslJobInfoParameter) –

  Is the parameter instance we check the available worker classes and there the job_info_class at each.

Returns:

• Type[Runner] –

  The selected runner class

Source code in src/qgis_server_light/worker/engine.py

def runner_plugin_by_job_info(self, job_info: QslJobInfoParameter) -> Type[Runner]:
    """
    Here we decide which plugin we load dynamically out of the available ones.

    Args:
        job_info: Is the parameter instance we check the available worker classes and there the
            job_info_class at each.

    Returns:
        The selected runner class
    """
    try:
        return self.available_runner_classes_by_job_info[
            job_info.__class__.__name__
        ]
    except KeyError:
        raise RuntimeError(f"Type {type(job_info)} not supported")

set_crashed()

Source code in src/qgis_server_light/worker/engine.py

def set_crashed(self):
    self.info.status = Status.CRASHED

set_processing()

Source code in src/qgis_server_light/worker/engine.py

def set_processing(self):
    self.info.status = Status.PROCESSING

set_waiting()

Source code in src/qgis_server_light/worker/engine.py

def set_waiting(self):
    self.info.status = Status.WAITING

EngineContext dataclass

Source code in src/qgis_server_light/worker/engine.py

@dataclass
class EngineContext:
    base_path: Union[str, pathlib.Path]

base_path: Union[str, pathlib.Path] instance-attribute

__init__(base_path: Union[str, pathlib.Path]) -> None

image_utils

qgis

CredentialsHelper

Bases: QgsCredentials

Source code in src/qgis_server_light/worker/qgis.py

class CredentialsHelper(QgsCredentials):
    def __init__(self):
        super().__init__()
        self.setInstance(self)

    def request(self, realm, username, password, message):
        logging.warning(message)
        return True, None, None

    def requestMasterPassword(self, password, stored):
        logging.warning("Master password requested")

__init__()

Source code in src/qgis_server_light/worker/qgis.py

def __init__(self):
    super().__init__()
    self.setInstance(self)

request(realm, username, password, message)

Source code in src/qgis_server_light/worker/qgis.py

def request(self, realm, username, password, message):
    logging.warning(message)
    return True, None, None

requestMasterPassword(password, stored)

Source code in src/qgis_server_light/worker/qgis.py

def requestMasterPassword(self, password, stored):
    logging.warning("Master password requested")

Qgis(svg_paths: Optional[List[str]], log_level)

Source code in src/qgis_server_light/worker/qgis.py

def Qgis(svg_paths: Optional[List[str]], log_level):
    os.environ["QT_QPA_PLATFORM"] = "offscreen"
    qgs = QgsApplication([], False)
    qgs.initQgis()
    if svg_paths:
        _svg_paths = qgs.svgPaths()
        # we do fast set algebra to always have unique list of paths
        # https://docs.python.org/3/library/stdtypes.html#frozenset.union
        qgs.setSvgPaths(list(set(_svg_paths) | set(svg_paths)))
    logging.debug(f"Application Path: {qgs.prefixPath()}")
    logging.info(f"QGIS Version {Qgis_.version()}")

    if log_level == logging.DEBUG:
        logging.debug("QGIS Debugging enabled")

        def write_log_message(message, tag, level):
            logging.debug(f"{tag}({level}): {message}")

        QgsApplication.messageLog().messageReceived.connect(write_log_message)

        qgs.credentialsHelper = CredentialsHelper()

    return qgs

version() -> int

Source code in src/qgis_server_light/worker/qgis.py

def version() -> int:
    return Qgis_.versionInt()

version_name() -> str

Source code in src/qgis_server_light/worker/qgis.py

def version_name() -> str:
    return Qgis_.version()

qgis_type_serializer

QDateConverter

Bases: Converter

Source code in src/qgis_server_light/worker/qgis_type_serializer.py

class QDateConverter(Converter):
    format = "yyyy-MM-dd"

    def deserialize(self, value: str, **kwargs: Any) -> QDate:
        return QDate.fromString(value, self.format)

    def serialize(self, value: QDate, **kwargs: Any) -> Optional[str]:
        if value:
            return value.toString(self.format)
        else:
            return None

format = 'yyyy-MM-dd' class-attribute instance-attribute

deserialize(value: str, **kwargs: Any) -> QDate

Source code in src/qgis_server_light/worker/qgis_type_serializer.py

def deserialize(self, value: str, **kwargs: Any) -> QDate:
    return QDate.fromString(value, self.format)

serialize(value: QDate, **kwargs: Any) -> Optional[str]

Source code in src/qgis_server_light/worker/qgis_type_serializer.py

def serialize(self, value: QDate, **kwargs: Any) -> Optional[str]:
    if value:
        return value.toString(self.format)
    else:
        return None

register_converters_at_runtime()

Source code in src/qgis_server_light/worker/qgis_type_serializer.py

@contextmanager
def register_converters_at_runtime():
    def register(custom_type, converter_instance, registered_types):
        converter.register_converter(custom_type, converter_instance)
        registered_types.append(custom_type)

    registered = []
    try:
        register(QDate, QDateConverter(), registered)
        # register further types here
        yield
    finally:
        for tp in registered:
            try:
                converter.unregister_converter(tp)
            except KeyError:
                pass

redis

DEFAULT_DATA_ROOT = '/io/data' module-attribute

DEFAULT_SVG_PATH = '/io/svg' module-attribute

RedisEngine

Bases: Engine

Source code in src/qgis_server_light/worker/redis.py

class RedisEngine(Engine):
    def __init__(
        self,
        context: EngineContext,
        runner_plugins: list[str],
        svg_paths: Optional[List] = None,
    ) -> None:
        self.boot_start = time.time()
        super().__init__(context, runner_plugins, svg_paths)
        self.shutdown = False
        self.retry_wait = 0.01
        self.max_retries = 11
        self.info_expire: int = 300

    def retry_handling_with_jitter(self, count: int):
        if count <= self.max_retries:
            sleep = math.pow(2, count) * self.retry_wait
            logging.warning(f"Retrying in {sleep} seconds...")
            time.sleep(sleep)
        else:
            self.exit_connection_error()

    @staticmethod
    def exit_connection_error():
        logging.error("Shutting down => now connection to Redis")
        exit(404)

    def exit_gracefully(self, signum, frame):
        logging.error(f"Received: {signum}")
        self.shutdown = True
        exit(0)

    @staticmethod
    def set_job_runtime_status(
        job_id: str,
        pipeline: Pipeline,
        status: str,
        start_time: float,
    ):
        duration = time.time() - start_time
        ts = datetime.datetime.now().isoformat()
        pipeline.hset(f"job:{job_id}", RedisQueue.job_status_key, status)
        pipeline.hset(
            f"job:{job_id}",
            f"{RedisQueue.job_timestamp_key}.{status}",
            ts,
        )
        pipeline.hset(f"job:{job_id}", RedisQueue.job_last_update_key, ts)
        pipeline.hset(f"job:{job_id}", RedisQueue.job_duration_key, str(duration))
        pipeline.execute()

    def heartbeat(self, client: Redis) -> datetime.datetime:
        now = datetime.datetime.now()
        client.hset(f"worker:{self.info.id}", "last_seen", now.isoformat())
        return now

    def register_worker(self, client: Redis):
        # writing worker info to redis
        client.hset(
            f"worker:{self.info.id}", "info", JsonSerializer().render(self.info)
        )
        # set timer to automatically remove worker info from list
        client.expire(f"worker:{self.info.id}", self.info_expire)
        # add worker to list of workers in redis
        client.sadd("workers", self.info.id)
        self.heartbeat(client)
        logging.info("Worker was registered in Redis")

    def retry_connection(self, redis_url: str, count: int):
        logging.warning(f"Could not connect to redis on `{redis_url}`.")
        self.retry_handling_with_jitter(count)

    def start(self, redis_url) -> Redis:
        signal.signal(signal.SIGINT, self.exit_gracefully)
        signal.signal(signal.SIGTERM, self.exit_gracefully)
        r = Redis.from_url(
            redis_url, decode_responses=True, retry=Retry(ExponentialBackoff(), 0)
        )
        retry_count = 0
        while True:
            try:
                retry_count += 1
                logging.debug(f"Looking up redis: {redis_url}")
                r.ping()
            except RedisConnectionError as e:
                logging.debug(f"Connection on Redis not successful => {e}")
                self.retry_connection(redis_url, retry_count)
            else:
                break
        logging.info(f"Connection to redis on `{redis_url}`successful.")
        return r

    def run(self, redis_url):
        r = self.start(redis_url)
        p = r.pipeline()
        expire_limit = self.info_expire * 0.95
        retry_count = 0
        while not self.shutdown:
            try:
                self.register_worker(r)
                logging.debug("Waiting for jobs")
                self.set_waiting()
                # this is blocking the loop until a job is found in the redis
                # list/queue, if there is one we take it, we have a timeout here, to
                # renew the workers heartbeat in redis
                result = r.blpop([RedisQueue.job_queue_name], int(expire_limit))
                if result is None:
                    now = self.heartbeat(r)
                    logging.debug(
                        f"Worker heartbeat renewed in queue {now.isoformat()}"
                    )
                    r.expire(f"worker:{self.info.id}", self.info_expire)
                    continue
                else:
                    _, job_id = result
            except RedisConnectionError:
                retry_count += 1
                self.retry_connection(redis_url, retry_count)
                continue
            start_time = time.time()
            try:
                # we inform, that the job is running.
                self.set_job_runtime_status(job_id, p, Status.RUNNING.value, start_time)

                job_info_json = r.hget(f"job:{job_id}", RedisQueue.job_info_key)
                job_info_class_name = r.hget(
                    f"job:{job_id}", RedisQueue.job_info_type_key
                )
                job_info_class = self.available_job_info_classes[job_info_class_name]
                job_info = JsonParser().from_string(job_info_json, job_info_class)
                result: JobResult = self.process(job_info)
                result.worker_id = self.info.id
                result.worker_host_name = socket.gethostname()
                data = pickle.dumps(result)

                # we inform, that the job was finished successful
                self.set_job_runtime_status(job_id, p, Status.SUCCESS.value, start_time)

                # we publish the result to any subscribers
                p.publish(f"{RedisQueue.job_channel_name}:{job_id}", data)

            except Exception as e:
                # preparation of the result, containing error information
                result = JobResult(id=job_id, data=str(e), content_type="text")
                result.worker_id = self.info.id
                result.worker_host_name = socket.gethostname()
                data = pickle.dumps(result)

                # we inform, that the job has failed with errors
                # self.set_job_runtime_status(job_id, p, Status.FAILURE.value,
                # start_time)

                # we publish the result to any subscribers
                p.publish(f"{RedisQueue.job_channel_name}:{job_id}", data)

                # we provide error information to the logs
                logging.error(e, exc_info=True)
            finally:
                p.execute()
            logging.debug(f"Job duration: {time.time() - start_time}")
        exit(0)

boot_start = time.time() instance-attribute

info_expire: int = 300 instance-attribute

max_retries = 11 instance-attribute

retry_wait = 0.01 instance-attribute

shutdown = False instance-attribute

__init__(context: EngineContext, runner_plugins: list[str], svg_paths: Optional[List] = None) -> None

Source code in src/qgis_server_light/worker/redis.py

def __init__(
    self,
    context: EngineContext,
    runner_plugins: list[str],
    svg_paths: Optional[List] = None,
) -> None:
    self.boot_start = time.time()
    super().__init__(context, runner_plugins, svg_paths)
    self.shutdown = False
    self.retry_wait = 0.01
    self.max_retries = 11
    self.info_expire: int = 300

exit_connection_error() staticmethod

Source code in src/qgis_server_light/worker/redis.py

@staticmethod
def exit_connection_error():
    logging.error("Shutting down => now connection to Redis")
    exit(404)

exit_gracefully(signum, frame)

Source code in src/qgis_server_light/worker/redis.py

def exit_gracefully(self, signum, frame):
    logging.error(f"Received: {signum}")
    self.shutdown = True
    exit(0)

heartbeat(client: Redis) -> datetime.datetime

Source code in src/qgis_server_light/worker/redis.py

def heartbeat(self, client: Redis) -> datetime.datetime:
    now = datetime.datetime.now()
    client.hset(f"worker:{self.info.id}", "last_seen", now.isoformat())
    return now

register_worker(client: Redis)

Source code in src/qgis_server_light/worker/redis.py

def register_worker(self, client: Redis):
    # writing worker info to redis
    client.hset(
        f"worker:{self.info.id}", "info", JsonSerializer().render(self.info)
    )
    # set timer to automatically remove worker info from list
    client.expire(f"worker:{self.info.id}", self.info_expire)
    # add worker to list of workers in redis
    client.sadd("workers", self.info.id)
    self.heartbeat(client)
    logging.info("Worker was registered in Redis")

retry_connection(redis_url: str, count: int)

Source code in src/qgis_server_light/worker/redis.py

def retry_connection(self, redis_url: str, count: int):
    logging.warning(f"Could not connect to redis on `{redis_url}`.")
    self.retry_handling_with_jitter(count)

retry_handling_with_jitter(count: int)

Source code in src/qgis_server_light/worker/redis.py

def retry_handling_with_jitter(self, count: int):
    if count <= self.max_retries:
        sleep = math.pow(2, count) * self.retry_wait
        logging.warning(f"Retrying in {sleep} seconds...")
        time.sleep(sleep)
    else:
        self.exit_connection_error()

run(redis_url)

Source code in src/qgis_server_light/worker/redis.py

def run(self, redis_url):
    r = self.start(redis_url)
    p = r.pipeline()
    expire_limit = self.info_expire * 0.95
    retry_count = 0
    while not self.shutdown:
        try:
            self.register_worker(r)
            logging.debug("Waiting for jobs")
            self.set_waiting()
            # this is blocking the loop until a job is found in the redis
            # list/queue, if there is one we take it, we have a timeout here, to
            # renew the workers heartbeat in redis
            result = r.blpop([RedisQueue.job_queue_name], int(expire_limit))
            if result is None:
                now = self.heartbeat(r)
                logging.debug(
                    f"Worker heartbeat renewed in queue {now.isoformat()}"
                )
                r.expire(f"worker:{self.info.id}", self.info_expire)
                continue
            else:
                _, job_id = result
        except RedisConnectionError:
            retry_count += 1
            self.retry_connection(redis_url, retry_count)
            continue
        start_time = time.time()
        try:
            # we inform, that the job is running.
            self.set_job_runtime_status(job_id, p, Status.RUNNING.value, start_time)

            job_info_json = r.hget(f"job:{job_id}", RedisQueue.job_info_key)
            job_info_class_name = r.hget(
                f"job:{job_id}", RedisQueue.job_info_type_key
            )
            job_info_class = self.available_job_info_classes[job_info_class_name]
            job_info = JsonParser().from_string(job_info_json, job_info_class)
            result: JobResult = self.process(job_info)
            result.worker_id = self.info.id
            result.worker_host_name = socket.gethostname()
            data = pickle.dumps(result)

            # we inform, that the job was finished successful
            self.set_job_runtime_status(job_id, p, Status.SUCCESS.value, start_time)

            # we publish the result to any subscribers
            p.publish(f"{RedisQueue.job_channel_name}:{job_id}", data)

        except Exception as e:
            # preparation of the result, containing error information
            result = JobResult(id=job_id, data=str(e), content_type="text")
            result.worker_id = self.info.id
            result.worker_host_name = socket.gethostname()
            data = pickle.dumps(result)

            # we inform, that the job has failed with errors
            # self.set_job_runtime_status(job_id, p, Status.FAILURE.value,
            # start_time)

            # we publish the result to any subscribers
            p.publish(f"{RedisQueue.job_channel_name}:{job_id}", data)

            # we provide error information to the logs
            logging.error(e, exc_info=True)
        finally:
            p.execute()
        logging.debug(f"Job duration: {time.time() - start_time}")
    exit(0)

set_job_runtime_status(job_id: str, pipeline: Pipeline, status: str, start_time: float) staticmethod

Source code in src/qgis_server_light/worker/redis.py

@staticmethod
def set_job_runtime_status(
    job_id: str,
    pipeline: Pipeline,
    status: str,
    start_time: float,
):
    duration = time.time() - start_time
    ts = datetime.datetime.now().isoformat()
    pipeline.hset(f"job:{job_id}", RedisQueue.job_status_key, status)
    pipeline.hset(
        f"job:{job_id}",
        f"{RedisQueue.job_timestamp_key}.{status}",
        ts,
    )
    pipeline.hset(f"job:{job_id}", RedisQueue.job_last_update_key, ts)
    pipeline.hset(f"job:{job_id}", RedisQueue.job_duration_key, str(duration))
    pipeline.execute()

start(redis_url) -> Redis

Source code in src/qgis_server_light/worker/redis.py

def start(self, redis_url) -> Redis:
    signal.signal(signal.SIGINT, self.exit_gracefully)
    signal.signal(signal.SIGTERM, self.exit_gracefully)
    r = Redis.from_url(
        redis_url, decode_responses=True, retry=Retry(ExponentialBackoff(), 0)
    )
    retry_count = 0
    while True:
        try:
            retry_count += 1
            logging.debug(f"Looking up redis: {redis_url}")
            r.ping()
        except RedisConnectionError as e:
            logging.debug(f"Connection on Redis not successful => {e}")
            self.retry_connection(redis_url, retry_count)
        else:
            break
    logging.info(f"Connection to redis on `{redis_url}`successful.")
    return r

main() -> None

Source code in src/qgis_server_light/worker/redis.py

def main() -> None:
    parser = argparse.ArgumentParser()

    parser.add_argument("--redis-url", type=str, help="redis url")

    parser.add_argument(
        "--log-level",
        type=str,
        help="log level (debug, info, warning or error)",
        default="info",
    )

    parser.add_argument(
        "--data-root",
        type=str,
        help=f"Absolute path to the data dir. Defaults to {DEFAULT_DATA_ROOT}",
        default=DEFAULT_DATA_ROOT,
    )

    parser.add_argument(
        "--svg-path",
        type=str,
        help=f"Absolute path to additional svg files. Multiple paths "
        f"can be separated by `:`. Defaults to {DEFAULT_SVG_PATH}",
        default=DEFAULT_SVG_PATH,
    )

    args = parser.parse_args()

    logging.basicConfig(
        level=args.log_level.upper(), format="%(asctime)s [%(levelname)s] %(message)s"
    )

    if not args.redis_url:
        raise AssertionError(
            "no redis host specified: start qgis-server-light "
            "with '--redis-url <QSL_REDIS_URL>'"
        )

    svg_paths = args.svg_path.split(":")
    engine = RedisEngine(
        EngineContext(args.data_root),
        [
            "qgis_server_light.worker.runner.render.RenderRunner",
            "qgis_server_light.worker.runner.feature.GetFeatureRunner",
            # Not fully functional yet
            # "qgis_server_light.worker.runner.feature_info.GetFeatureInfoRunner",
        ],
        svg_paths=svg_paths,
    )
    engine.run(
        args.redis_url,
    )

runner

common

JobContext dataclass

Source code in src/qgis_server_light/worker/runner/common.py

@dataclass
class JobContext:
    base_path: str | Path

base_path: str | Path instance-attribute

__init__(base_path: str | Path) -> None

MapRunner

Bases: Runner

Base class for any runner that interacts with a map. Not runnable by itself.

Source code in src/qgis_server_light/worker/runner/common.py

class MapRunner(Runner):
    """Base class for any runner that interacts with a map.
    Not runnable by itself.
    """

    map_layers: List[QgsMapLayer]
    vector_layer_drivers = [
        "ogr",
        "postgres",
        "spatialite",
        "mssql",
        "oracle",
        "wfs",
        "delimitedtext",
        "gpx",
        "arcgisfeatureserver",
    ]
    raster_layer_drivers = [
        "gdal",
        "wms",
        "xyz",
        "arcgismapserver",
        "wcs",
    ]
    custom_layer_drivers = ["xyzvectortiles", "mbtilesvectortiles"]
    default_style_name = "default"

    def __init__(
        self,
        qgis: QgsApplication,
        context: JobContext,
        job_info: QslJobInfoParameter,
        layer_cache: Optional[Dict] = None,
    ) -> None:
        self.qgis = qgis
        self.context = context
        self.job_info = job_info
        self.map_layers = list()
        self.layer_cache = layer_cache

    def _get_map_settings(self, layers: List[QgsMapLayer]) -> QgsMapSettings:
        """Produces a QgsMapSettings object from a set of layers"""
        expression_context_scope = QgsExpressionContextScope()
        expression_context_scope.setVariable("map_id", str(uuid.uuid4()))
        expression_context = QgsExpressionContext()
        expression_context.appendScope(expression_context_scope)
        settings = QgsMapSettings()
        settings.setExpressionContext(expression_context)

        def preprocessor(path):
            return path

        settings.pathResolver().setPathPreprocessor(preprocessor)
        settings.setOutputSize(
            QSize(int(self.job_info.job.width), int(self.job_info.job.height))
        )
        if self.job_info.job.dpi:
            settings.setOutputDpi(self.job_info.job.dpi)
        minx, miny, maxx, maxy = self.job_info.job.bbox.to_2d_list()
        bbox = QgsRectangle(float(minx), float(miny), float(maxx), float(maxy))
        settings.setExtent(bbox)
        settings.setLayers(layers)
        settings.setBackgroundColor(QColor(Qt.transparent))
        crs = self.job_info.job.crs
        destination_crs = QgsCoordinateReferenceSystem.fromOgcWmsCrs(crs)
        settings.setDestinationCrs(destination_crs)
        return settings

    def _load_style(self, qgs_layer: QgsMapLayer, job_layer_definition: QslJobLayer):
        logging.info(
            f"Preparing job_layer_definition Style: {job_layer_definition.style.name}"
        )
        style_doc = QDomDocument()
        style_xml = zlib.decompress(
            urlsafe_b64decode(job_layer_definition.style.definition)
        )
        style_doc.setContent(style_xml)
        success, _ = qgs_layer.importNamedStyle(style_doc)

        logging.info(f" ✓ Style loaded: {success}")

    def get_cache_name(self, job_layer_definition: QslJobLayer) -> str:
        """Central method to decide which name is used in the cache to
        identify a layer.
        """
        return job_layer_definition.id

    def _decide_drivers(self, job_layer_definition: QslJobLayer) -> QgsMapLayer:
        """Decides which type of layer we are dealing with and delegates initialization
        to the right method.

        Args:
            job_layer_definition: The job_layer_definition containing all
                information to initialize a QgsMapLayer.
        Returns:
            The newly created layer.
        Raises:
            LookupError: When the driver is not in the expected ranges.
        """
        if job_layer_definition.driver in self.vector_layer_drivers:
            qgs_layer = self._prepare_vector_layer(job_layer_definition)
        elif job_layer_definition.driver in self.raster_layer_drivers:
            qgs_layer = self._prepare_raster_layer(job_layer_definition)
        elif job_layer_definition.driver in self.custom_layer_drivers:
            qgs_layer = self._prepare_custom_layer(job_layer_definition)
        else:
            raise LookupError(f"Type not implemented: {job_layer_definition}")
        return qgs_layer

    def _handle_layer_cache(self, job_layer_definition: QslJobLayer) -> QgsMapLayer:
        """Checks if layer can be fetched directly from the cache or initiates the
        creation of a new layer otherwise.

        Args:
            job_layer_definition: The job_layer_definition containing all
                information to initialize a QgsMapLayer.
        Returns:
            The layer (from cache or newly created).
        """
        cache_name = self.get_cache_name(job_layer_definition)
        if self.layer_cache is not None and cache_name in self.layer_cache:
            logging.debug(
                f"Using cached job_layer_definition {job_layer_definition.name} (identifier: {cache_name})"
            )
            qgs_layer = self.layer_cache[cache_name]
        else:
            qgs_layer = self._decide_drivers(job_layer_definition)
            if qgs_layer.isValid():
                logging.debug(
                    f"Newly initialized layer {job_layer_definition.name} is valid: {qgs_layer.isValid()}"
                )
                if self.layer_cache is not None:
                    self.layer_cache[cache_name] = qgs_layer
            else:
                logging.error(qgs_layer.error().message())
                logging.error(qgs_layer.dataProvider().error().message())
                raise RuntimeError(
                    f"Newly initialized layer {job_layer_definition.name} is not valid. JobLayerDefinition: {job_layer_definition}"
                )
        return qgs_layer

    def _provide_layer(self, job_layer_definition: QslJobLayer) -> None:
        """Fetches the QGIS layer relevant for the requested job layer.

        Args:
            job_layer_definition: The job_layer_definition containing all
                information to initialize a QgsMapLayer.
        Returns:
            None
        """
        qgs_layer = self._handle_layer_cache(job_layer_definition)
        # applying the style to the job_layer_definition
        self._load_style(qgs_layer, job_layer_definition)
        self.map_layers.append(qgs_layer)

    def _handle_datasource_definition(self, job_layer_definition: QslJobLayer) -> dict:
        layer_source = json.loads(job_layer_definition.source)
        if not job_layer_definition.remote:
            # we make the relative path an absolute one with the configured base path
            layer_source["path"] = os.path.join(
                self.context.base_path,
                job_layer_definition.folder_name,
                layer_source["path"],
            )
        return layer_source

    def _decoded_layer_source_to_connection_string(
        self, driver: str, layer_source: dict
    ) -> str:
        return QgsProviderRegistry.instance().encodeUri(driver, layer_source)

    def _prepare_vector_layer(
        self, job_layer_definition: QslJobLayer
    ) -> QgsVectorLayer:
        """
        Initializes a QgsVectorLayer from a job_layer_definition.
        Args:
            job_layer_definition: The job_layer_definition definition as
                received from the runner.

        Returns:
            The QgsVectorLayer instance in case initialization went correctly.
        Raises:
            RuntimeError: In case the initialized job_layer_definition was not
                valid from QGIS point of view (mostly related to not available
                data sources).
        """

        layer_source = self._handle_datasource_definition(job_layer_definition)
        layer_source_path = self._decoded_layer_source_to_connection_string(
            job_layer_definition.driver, layer_source
        )

        # removed loadDefaultStyle=False because it seems to have no effect anymore
        options = QgsVectorLayer.LayerOptions(readExtentFromXml=False)
        options.skipCrValidation = True
        options.forceReadOnly = True

        qgs_layer = QgsVectorLayer(
            layer_source_path,
            job_layer_definition.name,
            job_layer_definition.driver,
            options,
        )
        if job_layer_definition.filter:
            if isinstance(job_layer_definition.filter, OgcFilter110):
                # TODO: This is potentially bad: We always get all features from datasource. However, QGIS
                #   does not seem to support sliding window feature filter out of the box...
                logging.info(" QslJobLayer is filtered by:")
                logging.info(job_layer_definition.filter.definition)
                filter_doc = QDomDocument()
                filter_doc.setContent(job_layer_definition.filter.definition)
                filter_expression = QgsOgcUtils.expressionFromOgcFilter(
                    filter_doc.documentElement(),
                    QgsOgcUtils.FilterVersion.FILTER_OGC_1_1,
                    qgs_layer,
                )
                existing_expression = qgs_layer.subsetString()
                if existing_expression:
                    # Combining with AND the originally defined expression always takes precedence
                    expression = f"({existing_expression}) AND ({filter_expression.expression()})"
                else:
                    expression = filter_expression.expression()
                qgs_layer.setSubsetString(expression)
        return qgs_layer

    def _prepare_custom_layer(
        self, job_layer_definition: QslJobLayer
    ) -> QgsVectorTileLayer:
        """Initializes a custom job_layer_definition"""
        layer_source = self._handle_datasource_definition(job_layer_definition)
        layer_source_path = self._decoded_layer_source_to_connection_string(
            job_layer_definition.driver, layer_source
        )
        qgs_layer = QgsVectorTileLayer(layer_source_path, job_layer_definition.name)
        return qgs_layer

    def _prepare_raster_layer(
        self, job_layer_definition: QslJobLayer
    ) -> QgsRasterLayer:
        """Initializes a raster job_layer_definition"""
        layer_source = self._handle_datasource_definition(job_layer_definition)
        layer_source_path = self._decoded_layer_source_to_connection_string(
            job_layer_definition.driver, layer_source
        )
        qgs_layer = QgsRasterLayer(
            layer_source_path,
            job_layer_definition.name,
            job_layer_definition.driver,
        )
        return qgs_layer

context = context instance-attribute

custom_layer_drivers = ['xyzvectortiles', 'mbtilesvectortiles'] class-attribute instance-attribute

default_style_name = 'default' class-attribute instance-attribute

job_info = job_info instance-attribute

layer_cache = layer_cache instance-attribute

map_layers: List[QgsMapLayer] = list() instance-attribute

qgis = qgis instance-attribute

raster_layer_drivers = ['gdal', 'wms', 'xyz', 'arcgismapserver', 'wcs'] class-attribute instance-attribute

vector_layer_drivers = ['ogr', 'postgres', 'spatialite', 'mssql', 'oracle', 'wfs', 'delimitedtext', 'gpx', 'arcgisfeatureserver'] class-attribute instance-attribute

__init__(qgis: QgsApplication, context: JobContext, job_info: QslJobInfoParameter, layer_cache: Optional[Dict] = None) -> None

Source code in src/qgis_server_light/worker/runner/common.py

def __init__(
    self,
    qgis: QgsApplication,
    context: JobContext,
    job_info: QslJobInfoParameter,
    layer_cache: Optional[Dict] = None,
) -> None:
    self.qgis = qgis
    self.context = context
    self.job_info = job_info
    self.map_layers = list()
    self.layer_cache = layer_cache

get_cache_name(job_layer_definition: QslJobLayer) -> str

Central method to decide which name is used in the cache to identify a layer.

Source code in src/qgis_server_light/worker/runner/common.py

def get_cache_name(self, job_layer_definition: QslJobLayer) -> str:
    """Central method to decide which name is used in the cache to
    identify a layer.
    """
    return job_layer_definition.id

Runner

Bases: ABC

Source code in src/qgis_server_light/worker/runner/common.py

class Runner(ABC):
    job_info_class: Type[QslJobInfoParameter]

    def __init__(
        self,
        qgis: QgsApplication,
        context: JobContext,
        job_info: QslJobInfoParameter,
        layer_cache: Optional[Dict],
    ):
        # This is an abstract base class which is not runnable itself
        raise NotImplementedError()

    def run(self):
        # This is an abstract base class which is not runnable itself
        raise NotImplementedError()

    @classmethod
    def deserialize_job_info(cls, job_info: bytes):
        return JsonParser().from_bytes(job_info, cls.job_info_class)

job_info_class: Type[QslJobInfoParameter] instance-attribute

__init__(qgis: QgsApplication, context: JobContext, job_info: QslJobInfoParameter, layer_cache: Optional[Dict])

Source code in src/qgis_server_light/worker/runner/common.py

def __init__(
    self,
    qgis: QgsApplication,
    context: JobContext,
    job_info: QslJobInfoParameter,
    layer_cache: Optional[Dict],
):
    # This is an abstract base class which is not runnable itself
    raise NotImplementedError()

deserialize_job_info(job_info: bytes) classmethod

Source code in src/qgis_server_light/worker/runner/common.py

@classmethod
def deserialize_job_info(cls, job_info: bytes):
    return JsonParser().from_bytes(job_info, cls.job_info_class)

run()

Source code in src/qgis_server_light/worker/runner/common.py

def run(self):
    # This is an abstract base class which is not runnable itself
    raise NotImplementedError()

feature

GetFeatureRunner

Bases: MapRunner

Source code in src/qgis_server_light/worker/runner/feature.py

class GetFeatureRunner(MapRunner):
    job_info_class = QslJobInfoFeature

    def __init__(
        self,
        qgis: QgsApplication,
        context: JobContext,
        job_info: QslJobInfoFeature,
        layer_cache: Optional[Dict] = None,
        layer_style_cache: Optional[set] = None,
    ) -> None:
        super().__init__(qgis, context, job_info, layer_cache)

    def _clean_attribute(self, attribute_value: Any, idx: int, layer: QgsVectorLayer):
        if attribute_value == NULL:
            return None
        return attribute_value

    def _clean_attributes(self, attributes, layer):
        return [
            self._clean_attribute(attr, idx, layer)
            for idx, attr in enumerate(attributes)
        ]

    def _load_style(self, qgs_layer: QgsMapLayer, job_layer_definition: QslJobLayer):
        logging.info(" ✓ Omit style loading on WFS layer operation.")

    def run(self):
        query_collection = QueryCollection()
        numbers_matched = 0
        for query in self.job_info.job.queries:
            # we need to reset this because we want always only the layers related to the current query
            self.map_layers = []
            wfs_filter = query.filter
            for job_layer_definition in query.layers:
                self._provide_layer(job_layer_definition)

            for layer in self.map_layers:
                feature_collection = FeatureCollection(layer.name())
                query_collection.feature_collections.append(feature_collection)
                if isinstance(layer, QgsVectorLayer):
                    if wfs_filter is not None and wfs_filter.definition is not None:
                        # TODO: This is potentially bad: We always get all features from datasource. However, QGIS
                        #   does not seem to support sliding window feature filter out of the box...
                        logging.info(" QslJobLayer is filtered by:")
                        logging.info(f" {wfs_filter.definition}")
                        filter_doc = QDomDocument()
                        filter_doc.setContent(wfs_filter.definition)
                        # This is not correct in the WFS 2.0 way. We apply a filter to a job_layer_definition. But WFS 2.0
                        # allows filters on multiple layers.
                        expression = QgsOgcUtils.expressionFromOgcFilter(
                            filter_doc.documentElement(),
                            QgsOgcUtils.FilterVersion.FILTER_FES_2_0,
                        )
                        logging.info(
                            f" This was transformed to the QGIS expression (valid: {expression.isValid()})"
                        )
                        logging.info(f" '{expression.dump()}'")
                        feature_request = QgsFeatureRequest(expression)
                    else:
                        feature_request = QgsFeatureRequest()
                    layer_features = list(layer.getFeatures(feature_request))
                    numbers_matched += len(layer_features)
                    logging.info(f" Found {len(layer_features)} features")
                    if self.job_info.job.count:
                        layer_features = layer_features[
                            self.job_info.job.start_index : self.job_info.job.start_index
                            + self.job_info.job.count
                        ]
                    for layer_feature in layer_features:
                        property_list = zip(
                            layer_feature.fields().names(),
                            self._clean_attributes(layer_feature.attributes(), layer),
                        )
                        feature = Feature(
                            geometry=Geometry(
                                value=bytes(layer_feature.geometry().asWkb()),
                            )
                        )
                        feature_collection.features.append(feature)
                        for name, value in property_list:
                            feature.attributes.append(Attribute(name=name, value=value))
                else:
                    raise RuntimeError(
                        f"QslJobLayer type `{layer.type().name}` of layer `{layer.shortName()}` not supported by GetFeatureInfo"
                    )
        if numbers_matched > 0:
            query_collection.numbers_matched = numbers_matched
        with register_converters_at_runtime():
            data = JsonSerializer().render(query_collection).encode()
            return JobResult(
                id=self.job_info.id,
                data=data,
                content_type="application/qgis-server-light.interface.qgis.QueryCollection",
            )

job_info_class = QslJobInfoFeature class-attribute instance-attribute

__init__(qgis: QgsApplication, context: JobContext, job_info: QslJobInfoFeature, layer_cache: Optional[Dict] = None, layer_style_cache: Optional[set] = None) -> None

Source code in src/qgis_server_light/worker/runner/feature.py

def __init__(
    self,
    qgis: QgsApplication,
    context: JobContext,
    job_info: QslJobInfoFeature,
    layer_cache: Optional[Dict] = None,
    layer_style_cache: Optional[set] = None,
) -> None:
    super().__init__(qgis, context, job_info, layer_cache)

run()

Source code in src/qgis_server_light/worker/runner/feature.py

def run(self):
    query_collection = QueryCollection()
    numbers_matched = 0
    for query in self.job_info.job.queries:
        # we need to reset this because we want always only the layers related to the current query
        self.map_layers = []
        wfs_filter = query.filter
        for job_layer_definition in query.layers:
            self._provide_layer(job_layer_definition)

        for layer in self.map_layers:
            feature_collection = FeatureCollection(layer.name())
            query_collection.feature_collections.append(feature_collection)
            if isinstance(layer, QgsVectorLayer):
                if wfs_filter is not None and wfs_filter.definition is not None:
                    # TODO: This is potentially bad: We always get all features from datasource. However, QGIS
                    #   does not seem to support sliding window feature filter out of the box...
                    logging.info(" QslJobLayer is filtered by:")
                    logging.info(f" {wfs_filter.definition}")
                    filter_doc = QDomDocument()
                    filter_doc.setContent(wfs_filter.definition)
                    # This is not correct in the WFS 2.0 way. We apply a filter to a job_layer_definition. But WFS 2.0
                    # allows filters on multiple layers.
                    expression = QgsOgcUtils.expressionFromOgcFilter(
                        filter_doc.documentElement(),
                        QgsOgcUtils.FilterVersion.FILTER_FES_2_0,
                    )
                    logging.info(
                        f" This was transformed to the QGIS expression (valid: {expression.isValid()})"
                    )
                    logging.info(f" '{expression.dump()}'")
                    feature_request = QgsFeatureRequest(expression)
                else:
                    feature_request = QgsFeatureRequest()
                layer_features = list(layer.getFeatures(feature_request))
                numbers_matched += len(layer_features)
                logging.info(f" Found {len(layer_features)} features")
                if self.job_info.job.count:
                    layer_features = layer_features[
                        self.job_info.job.start_index : self.job_info.job.start_index
                        + self.job_info.job.count
                    ]
                for layer_feature in layer_features:
                    property_list = zip(
                        layer_feature.fields().names(),
                        self._clean_attributes(layer_feature.attributes(), layer),
                    )
                    feature = Feature(
                        geometry=Geometry(
                            value=bytes(layer_feature.geometry().asWkb()),
                        )
                    )
                    feature_collection.features.append(feature)
                    for name, value in property_list:
                        feature.attributes.append(Attribute(name=name, value=value))
            else:
                raise RuntimeError(
                    f"QslJobLayer type `{layer.type().name}` of layer `{layer.shortName()}` not supported by GetFeatureInfo"
                )
    if numbers_matched > 0:
        query_collection.numbers_matched = numbers_matched
    with register_converters_at_runtime():
        data = JsonSerializer().render(query_collection).encode()
        return JobResult(
            id=self.job_info.id,
            data=data,
            content_type="application/qgis-server-light.interface.qgis.QueryCollection",
        )

feature_info

GetFeatureInfoRunner

Bases: MapRunner

Source code in src/qgis_server_light/worker/runner/feature_info.py

class GetFeatureInfoRunner(MapRunner):
    job_info_class = QslJobInfoFeatureInfo

    def __init__(
        self,
        qgis: QgsApplication,
        context: JobContext,
        job_info: QslJobInfoFeatureInfo,
        layer_cache: Optional[Dict] = None,
    ) -> None:
        super().__init__(qgis, context, job_info, layer_cache)

    def _clean_attribute(self, attribute, idx, layer):
        if attribute == NULL:
            return None
        setup = layer.editorWidgetSetup(idx)
        fieldFormatter = QgsApplication.fieldFormatterRegistry().fieldFormatter(
            setup.type()
        )
        return fieldFormatter.representValue(
            layer, idx, setup.config(), None, attribute
        )

    def _clean_attributes(self, attributes, layer):
        return [
            self._clean_attribute(attr, idx, layer)
            for idx, attr in enumerate(attributes)
        ]

    def run(self):
        for job_layer_definition in self.job_info.job.layers:
            self._provide_layer(job_layer_definition)
        map_settings = self._get_map_settings(self.map_layers)
        # Estimate queryable bbox (2mm)
        map_to_pixel = map_settings.mapToPixel()
        map_point = map_to_pixel.toMapCoordinates(
            self.job_info.job.x, self.job_info.job.y
        )
        # Create identifiable bbox in map coordinates, ±2mm
        tolerance = 0.002 * 39.37 * map_settings.outputDpi()
        tl = QgsPointXY(map_point.x() - tolerance, map_point.y() - tolerance)
        br = QgsPointXY(map_point.x() + tolerance, map_point.y() + tolerance)
        rect = QgsRectangle(tl, br)
        render_context = QgsRenderContext.fromMapSettings(map_settings)

        features = list()
        for layer in self.map_layers:
            renderer = layer.renderer().clone() if layer.renderer() else None
            if renderer:
                renderer.startRender(render_context, layer.fields())

            if layer.type() == QgsMapLayerType.VectorLayer:
                layer_rect = map_settings.mapToLayerCoordinates(layer, rect)
                request = (
                    QgsFeatureRequest()
                    .setFilterRect(layer_rect)
                    .setFlags(QgsFeatureRequest.ExactIntersect)
                )
                for feature in layer.getFeatures(request):
                    if renderer.willRenderFeature(feature, render_context):
                        properties = OrderedDict(
                            zip(
                                feature.fields().names(),
                                self._clean_attributes(feature.attributes(), layer),
                            )
                        )
                        features.append({"type": "Feature", "properties": properties})
            else:
                raise RuntimeError(
                    f"Layer type `{layer.type().name}` of layer `{layer.shortName()}` not supported by GetFeatureInfo"
                )
            if renderer:
                renderer.stopRender(render_context)

        featurecollection = {"features": features, "type": "FeatureCollection"}
        return JobResult(
            id=self.job_info.id,
            data=json.dumps(featurecollection).encode("utf-8"),
            content_type="application/json",
        )

job_info_class = QslJobInfoFeatureInfo class-attribute instance-attribute

__init__(qgis: QgsApplication, context: JobContext, job_info: QslJobInfoFeatureInfo, layer_cache: Optional[Dict] = None) -> None

Source code in src/qgis_server_light/worker/runner/feature_info.py

def __init__(
    self,
    qgis: QgsApplication,
    context: JobContext,
    job_info: QslJobInfoFeatureInfo,
    layer_cache: Optional[Dict] = None,
) -> None:
    super().__init__(qgis, context, job_info, layer_cache)

run()

Source code in src/qgis_server_light/worker/runner/feature_info.py

def run(self):
    for job_layer_definition in self.job_info.job.layers:
        self._provide_layer(job_layer_definition)
    map_settings = self._get_map_settings(self.map_layers)
    # Estimate queryable bbox (2mm)
    map_to_pixel = map_settings.mapToPixel()
    map_point = map_to_pixel.toMapCoordinates(
        self.job_info.job.x, self.job_info.job.y
    )
    # Create identifiable bbox in map coordinates, ±2mm
    tolerance = 0.002 * 39.37 * map_settings.outputDpi()
    tl = QgsPointXY(map_point.x() - tolerance, map_point.y() - tolerance)
    br = QgsPointXY(map_point.x() + tolerance, map_point.y() + tolerance)
    rect = QgsRectangle(tl, br)
    render_context = QgsRenderContext.fromMapSettings(map_settings)

    features = list()
    for layer in self.map_layers:
        renderer = layer.renderer().clone() if layer.renderer() else None
        if renderer:
            renderer.startRender(render_context, layer.fields())

        if layer.type() == QgsMapLayerType.VectorLayer:
            layer_rect = map_settings.mapToLayerCoordinates(layer, rect)
            request = (
                QgsFeatureRequest()
                .setFilterRect(layer_rect)
                .setFlags(QgsFeatureRequest.ExactIntersect)
            )
            for feature in layer.getFeatures(request):
                if renderer.willRenderFeature(feature, render_context):
                    properties = OrderedDict(
                        zip(
                            feature.fields().names(),
                            self._clean_attributes(feature.attributes(), layer),
                        )
                    )
                    features.append({"type": "Feature", "properties": properties})
        else:
            raise RuntimeError(
                f"Layer type `{layer.type().name}` of layer `{layer.shortName()}` not supported by GetFeatureInfo"
            )
        if renderer:
            renderer.stopRender(render_context)

    featurecollection = {"features": features, "type": "FeatureCollection"}
    return JobResult(
        id=self.job_info.id,
        data=json.dumps(featurecollection).encode("utf-8"),
        content_type="application/json",
    )

legend

GetLegendRunner

Bases: MapRunner

Source code in src/qgis_server_light/worker/runner/legend.py

class GetLegendRunner(MapRunner):
    def __init__(self, qgis, context: JobContext, job_info: QslJobInfoLegend) -> None:
        super().__init__(qgis, context, job_info)

    def run(self):
        # TODO Implement ....
        raise NotImplementedError()

__init__(qgis, context: JobContext, job_info: QslJobInfoLegend) -> None

Source code in src/qgis_server_light/worker/runner/legend.py

def __init__(self, qgis, context: JobContext, job_info: QslJobInfoLegend) -> None:
    super().__init__(qgis, context, job_info)

run()

Source code in src/qgis_server_light/worker/runner/legend.py

def run(self):
    # TODO Implement ....
    raise NotImplementedError()

process

ProcessRunner

Bases: MapRunner

Source code in src/qgis_server_light/worker/runner/process.py

class ProcessRunner(MapRunner):
    def __init__(
        self,
        qgis: QgsApplication,
        context: JobContext,
        job_info: QslJobInfoParameter,
        layer_cache: Optional[dict],
    ):
        super().__init__(qgis, context, job_info, layer_cache)

        ProcessingAlgFactory()
        providers = QgsProviderRegistry.instance().pluginList().split("\n")
        logging.info("Found Providers:")
        for provider in providers:
            logging.info(f" - {provider}")

    def load_providers(self, qgis: Qgis):
        qgis.processingRegistry().addProvider(QgsNativeAlgorithms())
        qgis.processingRegistry().addProvider(QgsPdalAlgorithms())

__init__(qgis: QgsApplication, context: JobContext, job_info: QslJobInfoParameter, layer_cache: Optional[dict])

Source code in src/qgis_server_light/worker/runner/process.py

def __init__(
    self,
    qgis: QgsApplication,
    context: JobContext,
    job_info: QslJobInfoParameter,
    layer_cache: Optional[dict],
):
    super().__init__(qgis, context, job_info, layer_cache)

    ProcessingAlgFactory()
    providers = QgsProviderRegistry.instance().pluginList().split("\n")
    logging.info("Found Providers:")
    for provider in providers:
        logging.info(f" - {provider}")

load_providers(qgis: Qgis)

Source code in src/qgis_server_light/worker/runner/process.py

def load_providers(self, qgis: Qgis):
    qgis.processingRegistry().addProvider(QgsNativeAlgorithms())
    qgis.processingRegistry().addProvider(QgsPdalAlgorithms())

render

RenderRunner

Bases: MapRunner

Responsible for rendering a QslRenderJob to an image.

Source code in src/qgis_server_light/worker/runner/render.py

class RenderRunner(MapRunner):
    """Responsible for rendering a QslRenderJob to an image."""

    job_info_class = QslJobInfoRender

    def __init__(
        self,
        qgis: QgsApplication,
        context: JobContext,
        job_info: QslJobInfoRender,
        layer_cache: Optional[Dict] = None,
    ) -> None:
        super().__init__(qgis, context, job_info, layer_cache)

    @classmethod
    def image_formats(cls):
        return {"image/png": cls._encode_png, "image/jpeg": cls._encode_jpg}

    def run(self):
        """Run this runner.
        Returns:
            A JobResult with the content_type and image_data (bytes) of the rendered image.
        """
        logging.info(f"Executing job: {self.job_info}")
        feature_filter = QgsFeatureFilter()
        for job_layer_definition in self.job_info.job.layers:
            self._provide_layer(job_layer_definition)
        map_settings = self._get_map_settings(self.map_layers)
        filter_providers = QgsFeatureFilterProviderGroup()
        filter_providers.addProvider(feature_filter)
        renderer = QgsMapRendererParallelJob(map_settings)
        renderer.setFeatureFilterProvider(filter_providers)
        event_loop = QEventLoop(self.qgis)
        renderer.finished.connect(event_loop.quit)
        renderer.start()
        event_loop.exec_()
        img = renderer.renderedImage()
        img.setDotsPerMeterX(int(map_settings.outputDpi() * 39.37))
        img.setDotsPerMeterY(int(map_settings.outputDpi() * 39.37))
        content_type, image_data = self._encode_image(img, self.job_info.job.format)
        return JobResult(
            id=self.job_info.id, data=image_data, content_type=content_type
        )

    def _encode_image(self, image: QImage, fmt: str) -> Tuple[str, bytearray]:
        """Encodes an image in a specific mime type
        Args:
            image (QImage): The image to encode
            fmt (str): The mime type of the format
        Returns:
            A tuple with mime type and bytes-like object of an encoded image in the desired format
        """
        try:
            fmt = fmt.lower()
            encoding_method = self.image_formats()[fmt]
            return fmt, encoding_method(image)
        except KeyError:
            raise RuntimeError(
                f"Requested mimtype '{fmt}' was found in {list(self.image_formats.keys())}."
            )

    @staticmethod
    def _encode_png(image: QImage):
        image.convertTo(QImage.Format_RGBA8888)
        image_data = fpng_encode_image_to_memory(
            image.constBits().asstring(image.sizeInBytes()),
            image.width(),
            image.height(),
            0,
            CompressionFlags.NONE,
        )
        return image_data

    @staticmethod
    def _encode_jpg(image: QImage):
        image_data = QByteArray()
        buf = QBuffer(image_data)
        buf.open(QIODevice.WriteOnly)
        image.save(buf, "JPG")
        return image_data

job_info_class = QslJobInfoRender class-attribute instance-attribute

__init__(qgis: QgsApplication, context: JobContext, job_info: QslJobInfoRender, layer_cache: Optional[Dict] = None) -> None

Source code in src/qgis_server_light/worker/runner/render.py

def __init__(
    self,
    qgis: QgsApplication,
    context: JobContext,
    job_info: QslJobInfoRender,
    layer_cache: Optional[Dict] = None,
) -> None:
    super().__init__(qgis, context, job_info, layer_cache)

image_formats() classmethod

Source code in src/qgis_server_light/worker/runner/render.py

@classmethod
def image_formats(cls):
    return {"image/png": cls._encode_png, "image/jpeg": cls._encode_jpg}

run()

Run this runner. Returns: A JobResult with the content_type and image_data (bytes) of the rendered image.

Source code in src/qgis_server_light/worker/runner/render.py

def run(self):
    """Run this runner.
    Returns:
        A JobResult with the content_type and image_data (bytes) of the rendered image.
    """
    logging.info(f"Executing job: {self.job_info}")
    feature_filter = QgsFeatureFilter()
    for job_layer_definition in self.job_info.job.layers:
        self._provide_layer(job_layer_definition)
    map_settings = self._get_map_settings(self.map_layers)
    filter_providers = QgsFeatureFilterProviderGroup()
    filter_providers.addProvider(feature_filter)
    renderer = QgsMapRendererParallelJob(map_settings)
    renderer.setFeatureFilterProvider(filter_providers)
    event_loop = QEventLoop(self.qgis)
    renderer.finished.connect(event_loop.quit)
    renderer.start()
    event_loop.exec_()
    img = renderer.renderedImage()
    img.setDotsPerMeterX(int(map_settings.outputDpi() * 39.37))
    img.setDotsPerMeterY(int(map_settings.outputDpi() * 39.37))
    content_type, image_data = self._encode_image(img, self.job_info.job.format)
    return JobResult(
        id=self.job_info.id, data=image_data, content_type=content_type
    )