From 99848e28cf4562d06bc539c65e07e48add80c28f Mon Sep 17 00:00:00 2001
From: rldhont <rldhont@gmail.com>
Date: Tue, 19 Nov 2024 22:17:59 +0100
Subject: [PATCH] [Bugfix] Filter by polygon: table could be a query

The layer filtered by polygon could be a query `_features_ids_with_sql_query`

The polygon layer  could be a query `_polygon_for_groups_with_sql_query`
---
 lizmap_server/filter_by_polygon.py | 22 ++++++++++++++++------
 test/test_filter_by_polygon.py     | 17 +++++++++++++++++
 2 files changed, 33 insertions(+), 6 deletions(-)

diff --git a/lizmap_server/filter_by_polygon.py b/lizmap_server/filter_by_polygon.py
index 3db6c168..4f345d5c 100755
--- a/lizmap_server/filter_by_polygon.py
+++ b/lizmap_server/filter_by_polygon.py
@@ -282,6 +282,7 @@ def _polygon_for_groups_with_sql_query(self, groups_or_user: tuple) -> QgsGeomet
         Only for QGIS >= 3.10
         """
         uri = QgsDataSourceUri(self.polygon.source())
+
         try:
             sql = r"""
 WITH current_groups AS (
@@ -298,7 +299,7 @@ def _polygon_for_groups_with_sql_query(self, groups_or_user: tuple) -> QgsGeomet
 SELECT
         1 AS id, ST_AsBinary(ST_Union("{geom}")) AS geom
 FROM
-        "{schema}"."{table}" AS p,
+        {table_name} AS p,
         current_groups AS c
 WHERE
 c.user_group && (
@@ -318,8 +319,7 @@ def _polygon_for_groups_with_sql_query(self, groups_or_user: tuple) -> QgsGeomet
                 polygon_field=self.group_field,
                 groups_or_user=','.join(groups_or_user),
                 geom=uri.geometryColumn(),
-                schema=uri.schema(),
-                table=uri.table(),
+                table_name=FilterByPolygon._format_table_name(uri),
             )
             Logger.info(
                 f"Requesting the database about polygons for the current groups or user with : \n{sql}")
@@ -400,10 +400,9 @@ def _features_ids_with_sql_query(self, st_intersect: str) -> str:
         """
         uri = QgsDataSourceUri(self.layer.source())
 
-        sql = 'SELECT "{pk}" FROM "{schema}"."{table}" WHERE {st_intersect}'.format(
+        sql = 'SELECT "{pk}" FROM {table_name} WHERE {st_intersect}'.format(
             pk=self.primary_key,
-            schema=uri.schema(),
-            table=uri.table(),
+            table_name=FilterByPolygon._format_table_name(uri),
             st_intersect=st_intersect,
         )
         Logger.info(
@@ -495,3 +494,14 @@ def _format_qgis_expression_relationship(
     {current_geometry}
 )"""
         return expression
+
+    @classmethod
+    def _format_table_name(cls, uri: QgsDataSourceUri) -> str:
+        # is the datasource a query or a table ?
+        table_name = uri.table()
+        # is the datasource a query or a table ?
+        if not uri.schema() and table_name.startswith('(') and table_name.endswith(')'):
+            # it is a query
+            return table_name
+        # it is a table
+        return uri.quotedTablename()
diff --git a/test/test_filter_by_polygon.py b/test/test_filter_by_polygon.py
index e9c11183..b2a75b39 100644
--- a/test/test_filter_by_polygon.py
+++ b/test/test_filter_by_polygon.py
@@ -4,6 +4,7 @@
 
 from qgis.core import (
     QgsCoordinateReferenceSystem,
+    QgsDataSourceUri,
     QgsFeature,
     QgsGeometry,
     QgsProject,
@@ -280,3 +281,19 @@ def test_subset_string_postgres(self):
     ST_Centroid(\"geom\")
 )"""
         self.assertEqual(expected, sql)
+
+    def test_format_table_name(self):
+        """ Test format table name. """
+        uri = QgsDataSourceUri()
+        uri.setConnection("localhost", "5432", "dbname", "johny", "xxx")
+        uri.setDataSource("public", "roads", "the_geom", "cityid = 2643", "primary_key_field")
+
+        self.assertEqual('"public"."roads"', FilterByPolygon._format_table_name(uri))
+
+        uri.setDataSource("", "roads", "the_geom", "cityid = 2643", "primary_key_field")
+
+        self.assertEqual('"roads"', FilterByPolygon._format_table_name(uri))
+
+        uri.setDataSource("", '(SELECT * FROM "public"."roads")', "the_geom", "cityid = 2643", "primary_key_field")
+
+        self.assertEqual('(SELECT * FROM "public"."roads")', FilterByPolygon._format_table_name(uri))