Rererefactor names routes

apptax/taxonomie/routestaxref.py

@@ -5,7 +5,6 @@
 from sqlalchemy.orm.exc import NoResultFound
 from sqlalchemy.orm import raiseload, joinedload
 
-
 from ..utils.utilssqlalchemy import json_resp, serializeQuery, serializeQueryOneResult
 from .models import (
     Taxref,
@@ -236,37 +235,45 @@ def getTaxrefHierarchieBibNoms(rang):
 
 
 def genericTaxrefList(inBibtaxon, parameters):
-    taxrefColumns = Taxref.__table__.columns
-    bibNomsColumns = BibNoms.__table__.columns
-    q = db.session.query(Taxref, BibNoms.id_nom)
 
-    qcount = q.outerjoin(BibNoms, BibNoms.cd_nom == Taxref.cd_nom)
+    q = Taxref.query.options(raiseload("*"), joinedload(
+        Taxref.bib_nom).joinedload(BibNoms.listes))
+
+    nbResultsWithoutFilter = q.count()
+
+    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()
 
@@ -281,9 +288,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,
@@ -454,48 +486,3 @@ def get_bib_hab():
     data = db.session.query(BibTaxrefHabitats).all()
     return [d.as_dict() for d in data]
 
-
-@adresses.route("/names", methods=["GET"])
-@json_resp
-def get_names():
-    """
-    Route utilisée pour retourner les résultats de taxref
-
-    params GET (facultatifs):
-        - fields : liste des champs à retourner (liste des champs de VMTaxrefListForautocomplete + lists)
-        - id_liste : filtre sur une ou des listes associées à un taxon
-        - limit: nombre de résultats
-        - offset: numéro de la page
-
-    """
-    limit = request.args.get("limit", 20, int)
-    page = request.args.get("page", 1, int)
-
-    q = Taxref.query.options(raiseload("*"), joinedload("bib_nom"), joinedload("bib_nom.listes"))
-
-    id_liste = request.args.getlist("id_liste", None)
-
-    if id_liste and not id_liste == -1:
-        q = q.filter(CorNomListe.id_liste.in_(tuple(id_liste)))
-
-    # 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 "lists" in fields:
-            id_listes = []
-            if r.bib_nom:
-                id_listes = [l.id_liste for l in r.bib_nom[0].listes]
-            items.append(dict(data, lists=id_listes))
-        else:
-            items.append(data)
-
-    return items

apptax/tests/test_taxref.py

@@ -51,7 +51,8 @@ class TestAPITaxref:
                 "group2_inpn": str,
                 "group3_inpn": Or(None, str),
                 "url": Or(None, str),
-                "lists": [int],
+                "listes": [int],
+                "id_nom": Or(None, int),
             }
         ]
     )
@@ -127,34 +128,38 @@ def test_get_allnamebyListe_routes_without_list_filter_group3(self):
         if data:
             assert self.schema_allnamebyListe.is_valid(data)
 
-    def test_get_names_routes(self):
-        query_string = {"limit": 100}
-        response = self.client.get(url_for("taxref.get_names"), query_string=query_string)
+    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)
+            assert self.schema_names.is_valid(data['items'])
 
-    def test_get_names_routes_limit_fields(self):
-        query_string = {"limit": 10, "fields": "cd_nom,cd_ref,lists"}
-        response = self.client.get(url_for("taxref.get_names"), query_string=query_string)
-        current_schema = Schema([{"cd_nom": int, "cd_ref": int, "lists": [int]}])
+    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)
+            assert current_schema.is_valid(data['items'])
 
-    def test_get_names_routes_limit_filter_id_liste(self):
+    def test_getTaxrefList_routes_limit_filter_id_liste(self):
         query_string = {"id_liste": "100", "fields": "cd_nom"}
-        response = self.client.get(url_for("taxref.get_names"), query_string=query_string)
+        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)
-            assert len(data) == 9
+            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"))