Feat Synthese: add status, red lists filters

Resolve #1492.

backend/geonature/core/gn_synthese/utils/query_select_sqla.py

@@ -19,7 +19,9 @@
 from utils_flask_sqla_geo.utilsgeometry import circle_from_point
 
 from geonature.utils.env import DB
-from geonature.core.taxonomie.models import Taxref, CorTaxonAttribut, TaxrefLR
+from geonature.core.taxonomie.models import (
+    TaxrefLR
+)
 from geonature.core.gn_synthese.models import (
     Synthese,
     CorObserverSynthese,
@@ -32,7 +34,17 @@
     TDatasets,
 )
 from geonature.utils.errors import GeonatureApiError
-
+from geonature.core.ref_geo.models import (
+    CorAreaStatus,
+)
+from apptax.taxonomie.models import (
+    Taxref,
+    CorTaxonAttribut,
+    TaxrefBdcStatutTaxon,
+    TaxrefBdcStatutCorTextValues,
+    TaxrefBdcStatutText,
+    TaxrefBdcStatutValues,
+)
 
 class SyntheseQuery:
     """
@@ -331,6 +343,80 @@ def filter_other_filters(self):
             if colname.startswith("area"):
                 self.add_join(CorAreaSynthese, CorAreaSynthese.id_synthese, self.model.id_synthese)
                 self.query = self.query.where(CorAreaSynthese.id_area.in_(value))
+            elif colname.endswith("_red_lists"):
+                red_list_id = colname.replace("_red_lists", "")
+                all_red_lists_cfg = current_app.config["SYNTHESE"]["RED_LISTS_FILTERS"]
+                red_list_cfg = next((item for item in all_red_lists_cfg if item["id"] == red_list_id), None)
+                red_list_cte = (
+                    select([TaxrefBdcStatutTaxon.cd_ref, CorAreaStatus.id_area])
+                    .select_from(
+                        TaxrefBdcStatutTaxon.__table__
+                        .join(
+                            TaxrefBdcStatutCorTextValues, 
+                            TaxrefBdcStatutCorTextValues.id_value_text == TaxrefBdcStatutTaxon.id_value_text
+                        )
+                        .join(
+                            TaxrefBdcStatutText,
+                            TaxrefBdcStatutText.id_text == TaxrefBdcStatutCorTextValues.id_text
+                        )
+                        .join(
+                            TaxrefBdcStatutValues,
+                            TaxrefBdcStatutValues.id_value == TaxrefBdcStatutCorTextValues.id_value
+                        )
+                        .join(
+                            CorAreaStatus,
+                            CorAreaStatus.cd_sig == TaxrefBdcStatutText.cd_sig
+                        )
+                    )
+                    .where(TaxrefBdcStatutValues.code_statut.in_(value))
+                    .where(TaxrefBdcStatutText.cd_type_statut == red_list_cfg['status_type'])
+                    .where(TaxrefBdcStatutText.enable == True)
+                    .cte(name=f"{red_list_id}_red_list")
+                )
+                cas_red_list = aliased(CorAreaSynthese)
+                self.add_join(cas_red_list, cas_red_list.id_synthese, self.model.id_synthese)
+                self.add_join(Taxref, Taxref.cd_nom, self.model.cd_nom)
+                self.add_join_multiple_cond(red_list_cte, [
+                    red_list_cte.c.cd_ref == Taxref.cd_ref,
+                    red_list_cte.c.id_area == cas_red_list.id_area,
+                ])
+
+            elif colname.endswith("_status"):
+                status_id = colname.replace("_status", "")
+                all_status_cfg = current_app.config["SYNTHESE"]["STATUS_FILTERS"]
+                status_cfg = next((item for item in all_status_cfg if item["id"] == status_id), None)
+                # Check if a checkbox was used.
+                if (isinstance(value, list) and value[0] == True and len(status_cfg['status_types']) == 1):
+                    value = status_cfg['status_types']
+                status_cte = (
+                    select([TaxrefBdcStatutTaxon.cd_ref, CorAreaStatus.id_area])
+                    .select_from(
+                        TaxrefBdcStatutTaxon.__table__
+                        .join(
+                            TaxrefBdcStatutCorTextValues, 
+                            TaxrefBdcStatutCorTextValues.id_value_text == TaxrefBdcStatutTaxon.id_value_text
+                        )
+                        .join(
+                            TaxrefBdcStatutText,
+                            TaxrefBdcStatutText.id_text == TaxrefBdcStatutCorTextValues.id_text
+                        )
+                        .join(
+                            CorAreaStatus,
+                            CorAreaStatus.cd_sig == TaxrefBdcStatutText.cd_sig
+                        )
+                    )
+                    .where(TaxrefBdcStatutText.cd_type_statut.in_(value))
+                    .where(TaxrefBdcStatutText.enable == True)
+                    .distinct()
+                    .cte(name=f"{status_id}_status")
+                )
+                cas_status = aliased(CorAreaSynthese)
+                self.add_join(cas_status, cas_status.id_synthese, self.model.id_synthese)
+                self.add_join(Taxref, Taxref.cd_nom, self.model.cd_nom)
+                self.add_join_multiple_cond(status_cte, [
+                    status_cte.c.cd_ref == Taxref.cd_ref,
+                    status_cte.c.id_area == cas_status.id_area,
+                ])
             elif colname.startswith("id_"):
                 col = getattr(self.model.__table__.columns, colname)
                 self.query = self.query.where(col.in_(value))

backend/geonature/core/ref_geo/models.py

@@ -70,4 +70,12 @@ class LiMunicipalities(DB.Model):
 
     @hybrid_property
     def nom_com_dept(self):
-        return '{} ({})'.format(self.nom_com, self.insee_dep)
+        return '{} ({})'.format(self.nom_com, self.insee_dep)
+
+
+@serializable
+class CorAreaStatus(DB.Model):
+    __tablename__ = "cor_area_status"
+    __table_args__ = {"schema": "ref_geo"}
+    cd_sig = DB.Column(DB.Unicode, primary_key=True)
+    id_area = DB.Column(DB.Integer, ForeignKey("ref_geo.l_areas.id_area"), primary_key=True)