Improve routes /taxref

apptax/taxonomie/models.py

@@ -22,7 +22,7 @@ class BibNoms(db.Model):
     nom_francais = db.Column(db.Unicode)
     comments = db.Column(db.Unicode)
 
-    taxref = db.relationship("Taxref")
+    taxref = db.relationship("Taxref", back_populates="bib_nom")
     attributs = db.relationship("CorTaxonAttribut")
     listes = db.relationship("CorNomListe")
     # medias relationship defined through backref
@@ -125,6 +125,8 @@ class Taxref(db.Model):
     group3_inpn = db.Column(db.Unicode)
     url = db.Column(db.Unicode)
 
+    bib_nom = db.relationship("BibNoms", back_populates="taxref")
+
     @hybrid_property
     def nom_vern_or_lb_nom(self):
         return self.nom_vern if self.nom_vern else self.lb_nom
@@ -229,7 +231,7 @@ class VMTaxrefListForautocomplete(db.Model):
     __tablename__ = "vm_taxref_list_forautocomplete"
     __table_args__ = {"schema": "taxonomie"}
     gid = db.Column(db.Integer, primary_key=True)
-    cd_nom = db.Column(db.Integer)
+    cd_nom = db.Column(db.Integer, ForeignKey(Taxref.cd_nom))
     search_name = db.Column(db.Unicode)
     unaccent_search_name = db.Column(db.Unicode)
     cd_ref = db.Column(db.Integer)

apptax/taxonomie/routestaxref.py

@@ -3,7 +3,7 @@
 from flask import abort, jsonify, Blueprint, request
 from sqlalchemy import distinct, desc, func, and_
 from sqlalchemy.orm.exc import NoResultFound
-
+from sqlalchemy.orm import raiseload, joinedload
 
 from ..utils.utilssqlalchemy import json_resp, serializeQuery, serializeQueryOneResult
 from .models import (
@@ -235,37 +235,43 @@ def getTaxrefHierarchieBibNoms(rang):
 
 
 def genericTaxrefList(inBibtaxon, parameters):
-    taxrefColumns = Taxref.__table__.columns
-    bibNomsColumns = BibNoms.__table__.columns
-    q = db.session.query(Taxref, BibNoms.id_nom)
+    q = Taxref.query.options(raiseload("*"), joinedload(Taxref.bib_nom).joinedload(BibNoms.listes))
+
+    nbResultsWithoutFilter = q.count()
 
-    qcount = q.outerjoin(BibNoms, BibNoms.cd_nom == Taxref.cd_nom)
+    id_liste = parameters.getlist("id_liste", None)
 
-    nbResultsWithoutFilter = qcount.count()
+    if id_liste and not id_liste == -1:
+        from sqlalchemy.orm import aliased
+
+        filter_cor_nom_liste = aliased(CorNomListe)
+        filter_bib_noms = aliased(BibNoms)
+        q = q.join(filter_bib_noms, filter_bib_noms.cd_nom == Taxref.cd_nom)
+        q = q.join(filter_cor_nom_liste, filter_bib_noms.id_nom == filter_cor_nom_liste.id_nom)
+        q = q.filter(filter_cor_nom_liste.id_liste.in_(tuple(id_liste)))
 
     if inBibtaxon is True:
-        q = q.join(BibNoms, BibNoms.cd_nom == Taxref.cd_nom)
-    else:
-        q = q.outerjoin(BibNoms, BibNoms.cd_nom == Taxref.cd_nom)
+        q = q.filter(BibNoms.cd_nom.isnot(None))
 
     # Traitement des parametres
     limit = parameters.get("limit", 20, int)
     page = parameters.get("page", 1, int)
 
     for param in parameters:
-        if param in taxrefColumns and parameters[param] != "":
-            col = getattr(taxrefColumns, param)
+        if hasattr(Taxref, param) and parameters[param] != "":
+            col = getattr(Taxref, param)
             q = q.filter(col == parameters[param])
         elif param == "is_ref" and parameters[param] == "true":
             q = q.filter(Taxref.cd_nom == Taxref.cd_ref)
         elif param == "ilike":
             q = q.filter(Taxref.lb_nom.ilike(parameters[param] + "%"))
         elif param == "is_inbibtaxons" and parameters[param] == "true":
-            q = q.filter(bibNomsColumns.cd_nom.isnot(None))
+            q = q.filter(BibNoms.cd_nom.isnot(None))
         elif param.split("-")[0] == "ilike":
             value = unquote(parameters[param])
-            col = str(param.split("-")[1])
-            q = q.filter(taxrefColumns[col].ilike(value + "%"))
+            column = str(param.split("-")[1])
+            col = getattr(Taxref, column)
+            q = q.filter(col.ilike(value + "%"))
 
     nbResults = q.count()
 
@@ -280,9 +286,34 @@ def genericTaxrefList(inBibtaxon, parameters):
                 orderCol = orderCol.desc()
         q = q.order_by(orderCol)
 
+    # Filtrer champs demandés par la requête
+    fields = request.args.get("fields", type=str, default=[])
+    if fields:
+        fields = fields.split(",")
+    fields_to_filter = None
+    if fields:
+        fields_to_filter = [f for f in fields if getattr(Taxref, f, None)]
+
     results = q.paginate(page=page, per_page=limit, error_out=False)
+
+    items = []
+    for r in results.items:
+        data = r.as_dict(fields=fields_to_filter)
+        if not fields or "listes" in fields:
+            id_listes = []
+            if r.bib_nom:
+                id_listes = [l.id_liste for l in r.bib_nom[0].listes]
+            data = dict(data, listes=id_listes)
+        if not fields or "id_nom" in fields:
+            id_nom = None
+            if r.bib_nom:
+                id_nom = r.bib_nom[0].id_nom
+            data = dict(data, id_nom=id_nom)
+
+        items.append(data)
+
     return {
-        "items": [dict(d.Taxref.as_dict(), **{"id_nom": d.id_nom}) for d in results.items],
+        "items": items,
         "total": nbResultsWithoutFilter,
         "total_filtered": nbResults,
         "limit": limit,

apptax/tests/test_taxref.py

@@ -4,6 +4,8 @@
 from flask import url_for
 from schema import Schema, Optional, Or
 
+from .fixtures import attribut_example, noms_example
+
 
 @pytest.mark.usefixtures("client_class", "temporary_transaction")
 class TestAPITaxref:
@@ -23,7 +25,39 @@ class TestAPITaxref:
             }
         ]
     )
-
+    schema_names = Schema(
+        [
+            {
+                "cd_nom": int,
+                "id_statut": Or(None, str),
+                "id_habitat": Or(None, int),
+                "id_rang": str,
+                "phylum": Or(None, str),
+                "classe": Or(None, str),
+                "regne": Or(None, str),
+                "ordre": Or(None, str),
+                "famille": Or(None, str),
+                "sous_famille": Or(None, str),
+                "tribu": Or(None, str),
+                "cd_taxsup": Or(None, int),
+                "cd_sup": Or(None, int),
+                "cd_ref": int,
+                "lb_nom": str,
+                "lb_auteur": str,
+                "nom_complet": str,
+                "nom_complet_html": str,
+                "nom_vern": Or(None, str),
+                "nom_valide": str,
+                "nom_vern_eng": Or(None, str),
+                "group1_inpn": str,
+                "group2_inpn": str,
+                "group3_inpn": Or(None, str),
+                "url": Or(None, str),
+                "listes": [int],
+                "id_nom": Or(None, int),
+            }
+        ]
+    )
     schema_taxref_detail = Schema(
         {
             "cd_nom": int,
@@ -96,6 +130,35 @@ def test_get_allnamebyListe_routes_without_list_filter_group3(self):
         if data:
             assert self.schema_allnamebyListe.is_valid(data)
 
+    def test_getTaxrefList_routes(self):
+        query_string = {"limit": 10}
+        response = self.client.get(url_for("taxref.getTaxrefList"), query_string=query_string)
+        assert response.status_code == 200
+        data = response.json
+        if data:
+            assert self.schema_names.is_valid(data["items"])
+
+    def test_getTaxrefList_routes_limit_fields(self):
+        query_string = {"limit": 10, "fields": "cd_nom,cd_ref,listes"}
+        response = self.client.get(url_for("taxref.getTaxrefList"), query_string=query_string)
+        current_schema = Schema([{"cd_nom": int, "cd_ref": int, "listes": [int]}])
+        assert response.status_code == 200
+        data = response.json
+
+        if data:
+            assert current_schema.is_valid(data["items"])
+
+    def test_getTaxrefList_routes_limit_filter_id_liste(self, noms_example):
+        query_string = {"id_liste": 100, "fields": "cd_nom"}
+        response = self.client.get(url_for("taxref.getTaxrefList"), query_string=query_string)
+        current_schema = Schema([{"cd_nom": int}])
+        assert response.status_code == 200
+        data = response.json
+
+        if data:
+            assert current_schema.is_valid(data["items"])
+            assert len(data["items"]) == 9
+
     def test_get_distinct_routes(self):
         response = self.client.get(url_for("taxref.getDistinctField", field="regne"))
         assert response.status_code == 200