Feat/monitoring sites (PnX-SI#16)

* feat(api): wip began add site routes + tests

With site categories
Also add tests

* feat(api): add more routes

* test(api): add tests and fixtures

* style(api): applied black

* feat(db): add migration to remove id_module

Column in t_sites_groups

* refactor(api): move utils for routes from sites

* feat(api): wip: add sites groups route

* test(api): wip: begin adding fixture site_groups

* fix: remove id_module in all models

* chore: rename route for better consistency

* tests: moved site_groups in tests and add tests

* chore(api): applied black

* refactor(api): add filter params function

And refact routes to use it

backend/gn_module_monitoring/migrations/f24adb481f54_remove_id_module_from_sites_groups.py

@@ -0,0 +1,54 @@
+"""remove_id_module_from_sites_groups
+
+Revision ID: f24adb481f54
+Revises: 
+Create Date: 2022-12-13 16:00:00.512562
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+from gn_module_monitoring import MODULE_CODE
+
+# revision identifiers, used by Alembic.
+revision = "f24adb481f54"
+down_revision = "b53bafb13ce8"
+branch_labels = None
+depends_on = None
+
+monitorings_schema = "gn_monitoring"
+
+
+def upgrade():
+    op.drop_column("t_sites_groups", "id_module", schema=monitorings_schema)
+
+
+def downgrade():
+    op.add_column(
+        "t_sites_groups",
+        sa.Column(
+            "id_module",
+            sa.Integer(),
+            sa.ForeignKey(
+                f"gn_commons.t_modules.id_module",
+                name="fk_t_sites_groups_id_module",
+                ondelete="CASCADE",
+                onupdate="CASCADE",
+            ),
+            nullable=True,
+        ),
+        schema=monitorings_schema,
+    )
+    # Cannot use orm here because need the model to be "downgraded" as well
+    # Need to set nullable True above for existing rows
+    # FIXME: find a better way because need to assign a module...
+    statement = sa.text(
+        f"""
+         update {monitorings_schema}.t_sites_groups
+         set id_module = (select id_module 
+                          from gn_commons.t_modules tm 
+                          where module_code = '\:module_code');
+        """
+    )
+    op.execute(statement, module_code=MODULE_CODE)
+    op.alter_column("t_sites_groups", "id_module", nullable=False)

backend/gn_module_monitoring/monitoring/models.py

@@ -208,12 +208,6 @@ class TMonitoringSitesGroups(DB.Model):
         unique=True
     )
 
-    id_module = DB.Column(
-        DB.ForeignKey('gn_commons.t_modules.id_module'),
-        nullable=False,
-        unique=True
-    )
-
     uuid_sites_group = DB.Column(UUID(as_uuid=True), default=uuid4)
 
     sites_group_name = DB.Column(DB.Unicode)
@@ -287,21 +281,22 @@ class TMonitoringModules(TModules):
         lazy='joined'
     )
 
-    sites = DB.relationship(
-        'TMonitoringSites',
-        uselist=True,  # pourquoi pas par defaut ?
-        primaryjoin=TMonitoringSites.id_module == id_module,
-        foreign_keys=[id_module],
-        lazy="select",
-    )
+    # TODO: restore it with CorCategorySite
+    # sites = DB.relationship(
+    #     'TMonitoringSites',
+    #     uselist=True,  # pourquoi pas par defaut ?
+    #     primaryjoin=TMonitoringSites.id_module == id_module,
+    #     foreign_keys=[id_module],
+    #     lazy="select",
+    # )
 
-    sites_groups = DB.relationship(
-        'TMonitoringSitesGroups',
-        uselist=True,  # pourquoi pas par defaut ?
-        primaryjoin=TMonitoringSitesGroups.id_module == id_module,
-        foreign_keys=[id_module],
-        lazy="select",
-    )
+    # sites_groups = DB.relationship(
+    #     'TMonitoringSitesGroups',
+    #     uselist=True,  # pourquoi pas par defaut ?
+    #     primaryjoin=TMonitoringSitesGroups.id_module == id_module,
+    #     foreign_keys=[id_module],
+    #     lazy="select",
+    # )
 
     datasets = DB.relationship(
         'TDatasets',

backend/gn_module_monitoring/routes/site.py

@@ -0,0 +1,47 @@
+from typing import Tuple
+
+from flask import request
+from flask.json import jsonify
+from geonature.core.gn_monitoring.models import TBaseSites
+from werkzeug.datastructures import MultiDict
+
+from gn_module_monitoring.blueprint import blueprint
+from gn_module_monitoring.monitoring.models import BibCategorieSite
+from gn_module_monitoring.utils.routes import (filter_params, get_limit_offset,
+                                               paginate)
+
+
+@blueprint.route("/sites/categories", methods=["GET"])
+def get_categories():
+    params = MultiDict(request.args)
+    limit, page = get_limit_offset(params=params)
+
+    query = filter_params(query=BibCategorieSite.query, params=params)
+    query = query.order_by(BibCategorieSite.id_categorie)
+    return paginate(query=query, object_name="categories", limit=limit, page=page)
+
+
+@blueprint.route("/sites/categories/<int:id_categorie>", methods=["GET"])
+def get_categories_by_id(id_categorie):
+    query = BibCategorieSite.query.filter_by(id_categorie=id_categorie)
+    res = query.first()
+
+    return jsonify(res.as_dict())
+
+
+@blueprint.route("/sites", methods=["GET"])
+def get_sites():
+    params = MultiDict(request.args)
+    # TODO: add filter support
+    limit, page = get_limit_offset(params=params)
+    query = TBaseSites.query.join(
+        BibCategorieSite, TBaseSites.id_categorie == BibCategorieSite.id_categorie
+    )
+    query = filter_params(query=query, params=params)
+    return paginate(query=query, object_name="sites", limit=limit, page=page)
+
+
+@blueprint.route("/sites/module/<string:module_code>", methods=["GET"])
+def get_module_sites(module_code: str):
+    # TODO: load with site_categories.json API
+    return jsonify({"module_code": module_code})

backend/gn_module_monitoring/routes/sites_groups.py

@@ -0,0 +1,18 @@
+from flask import request
+from werkzeug.datastructures import MultiDict
+
+from gn_module_monitoring.blueprint import blueprint
+from gn_module_monitoring.monitoring.models import TMonitoringSitesGroups
+from gn_module_monitoring.utils.routes import (filter_params, get_limit_offset,
+                                               paginate)
+
+
+@blueprint.route("/sites_groups", methods=["GET"])
+def get_sites_groups():
+    params = MultiDict(request.args)
+    limit, page = get_limit_offset(params=params)
+
+    query = filter_params(query=TMonitoringSitesGroups.query, params=params)
+
+    query = query.order_by(TMonitoringSitesGroups.id_sites_group)
+    return paginate(query=query, object_name="sites_groups", limit=limit, page=page)

backend/gn_module_monitoring/tests/conftest.py

@@ -0,0 +1,2 @@
+from geonature.tests.fixtures import *
+from geonature.tests.fixtures import _session, app, users

backend/gn_module_monitoring/tests/fixtures/site.py

@@ -0,0 +1,45 @@
+import pytest
+from geoalchemy2.shape import from_shape
+from geonature.core.gn_monitoring.models import TBaseSites
+from geonature.utils.env import db
+from pypnnomenclature.models import BibNomenclaturesTypes, TNomenclatures
+from shapely.geometry import Point
+
+from gn_module_monitoring.monitoring.models import BibCategorieSite
+
+
+@pytest.fixture()
+def categories():
+    categories = [{"label": "gite", "config": {}}, {"label": "eolienne", "config": {}}]
+
+    categories = {cat["label"]: BibCategorieSite(**cat) for cat in categories}
+
+    with db.session.begin_nested():
+        db.session.add_all(categories.values())
+
+    return categories
+
+
+@pytest.fixture()
+def sites(users, categories):
+    user = users["user"]
+    geom_4326 = from_shape(Point(43, 24), srid=4326)
+    sites = {}
+    # TODO: get_nomenclature from label
+    site_type = TNomenclatures.query.filter(
+        BibNomenclaturesTypes.mnemonique == "TYPE_SITE", TNomenclatures.mnemonique == "Grotte"
+    ).one()
+    for i, key in enumerate(categories.keys()):
+        sites[key] = TBaseSites(
+            id_inventor=user.id_role,
+            id_digitiser=user.id_role,
+            base_site_name=f"Site{i}",
+            base_site_description=f"Description{i}",
+            base_site_code=f"Code{i}",
+            geom=geom_4326,
+            id_nomenclature_type_site=site_type.id_nomenclature,
+            id_categorie=categories[key].id_categorie,
+        )
+    with db.session.begin_nested():
+        db.session.add_all(sites.values())
+    return sites

backend/gn_module_monitoring/tests/fixtures/sites_groups.py

@@ -0,0 +1,16 @@
+import pytest
+from geonature.utils.env import db
+
+from gn_module_monitoring.monitoring.models import TMonitoringSitesGroups
+
+
+@pytest.fixture
+def sites_groups():
+    names = ["Site_eolien", "Site_Groupe"]
+
+    groups = {name: TMonitoringSitesGroups(sites_group_name=name) for name in names}
+
+    with db.session.begin_nested():
+        db.session.add_all(groups.values())
+
+    return groups

backend/gn_module_monitoring/tests/test_monitoring/test_routes/test_site.py

@@ -0,0 +1,41 @@
+import pytest
+from flask import url_for
+
+from gn_module_monitoring.tests.fixtures.site import categories, sites
+
+
+@pytest.mark.usefixtures("client_class", "temporary_transaction")
+class TestSite:
+    def test_get_categories_by_id(self, categories):
+        for cat in categories.values():
+            r = self.client.get(
+                url_for(
+                    "monitorings.get_categories_by_id",
+                    id_categorie=cat.id_categorie,
+                )
+            )
+            assert r.json["label"] == cat.label
+
+    def test_get_categories(self, categories):
+        r = self.client.get(url_for("monitorings.get_categories"))
+
+        assert r.json["count"] >= len(categories)
+        assert all([cat.as_dict() in r.json["categories"] for cat in categories.values()])
+
+    def test_get_categories_label(self, categories):
+        label = list(categories.keys())[0]
+
+        r = self.client.get(url_for("monitorings.get_categories"), query_string={"label": label})
+        assert categories[label].as_dict() in r.json["categories"]
+
+    def test_get_sites(self, sites):
+        r = self.client.get(url_for("monitorings.get_sites"))
+
+        assert r.json["count"] >= len(sites)
+        assert any([site.as_dict() in r.json["sites"] for site in sites.values()])
+
+    def test_get_module_sites(self):
+        module_code = "TEST"
+        r = self.client.get(url_for("monitorings.get_module_sites", module_code=module_code))
+
+        assert r.json["module_code"] == module_code

backend/gn_module_monitoring/tests/test_monitoring/test_routes/test_sites_groups.py

@@ -0,0 +1,25 @@
+import pytest
+from flask import url_for
+
+from gn_module_monitoring.tests.fixtures.sites_groups import sites_groups
+
+
+@pytest.mark.usefixtures("client_class", "temporary_transaction")
+class TestSitesGroups:
+    def test_get_sites_groups(self, sites_groups):
+        r = self.client.get(url_for("monitorings.get_sites_groups"))
+
+        assert r.json["count"] >= len(sites_groups)
+        assert all([group.as_dict() in r.json["sites_groups"] for group in sites_groups.values()])
+
+    def test_get_sites_groups_filter_name(self, sites_groups):
+        name, name_not_present = list(sites_groups.keys())
+
+        r = self.client.get(
+            url_for("monitorings.get_sites_groups"), query_string={"sites_group_name": name}
+        )
+
+        assert r.json["count"] >= 1
+        json_sites_groups = r.json["sites_groups"]
+        assert sites_groups[name].as_dict() in json_sites_groups
+        assert sites_groups[name_not_present].as_dict() not in json_sites_groups

backend/gn_module_monitoring/utils/routes.py

@@ -0,0 +1,26 @@
+from typing import Tuple
+
+from flask import Response
+from flask.json import jsonify
+from sqlalchemy.orm import Query
+from werkzeug.datastructures import MultiDict
+
+
+def get_limit_offset(params: MultiDict) -> Tuple[int]:
+    return params.pop("limit", 50), params.pop("offset", 1)
+
+
+def paginate(query: Query, object_name: str, limit: int, page: int) -> Response:
+    result = query.paginate(page=page, error_out=False, max_per_page=limit)
+    data = {
+        object_name: [res.as_dict() for res in result.items],
+        "count": result.total,
+        "limit": limit,
+        "offset": page - 1,
+    }
+    return jsonify(data)
+
+def filter_params(query: Query, params: MultiDict) -> Query: 
+    if len(params) != 0:
+        query = query.filter_by(**params)
+    return query