from flask import Blueprint, request, g
from geonature.core.gn_monitoring.schema import TIndividualsSchema
from geonature.core.gn_permissions.tools import get_scope
from marshmallow import ValidationError, EXCLUDE
from sqlalchemy.sql import func, select
from geojson import FeatureCollection
from werkzeug.exceptions import BadRequest, Forbidden, NotFound
from geonature.core.gn_commons.models import TModules
from geonature.core.gn_permissions.decorators import _forbidden_message, login_required
from geonature.utils.env import DB
from ref_geo.models import LAreas
from sqlalchemy import select
from sqlalchemy.sql import func
from geonature.core.gn_monitoring.models import (
TBaseSites,
TIndividuals,
cor_site_area,
cor_site_module,
)
from utils_flask_sqla.response import json_resp
from utils_flask_sqla_geo.generic import get_geojson_feature
[docs]
routes = Blueprint("gn_monitoring", __name__)
@routes.route("/siteslist", methods=["GET"])
@json_resp
[docs]
def get_list_sites():
"""
Return the sites list for an application in a dict {id_base_site, nom site}
.. :quickref: Monitoring;
:param id_base_site: id of base site
:param module_code: code of the module
:param id_module: id of the module
:param base_site_name: part of the name of the site
:param type: int
"""
query = select(TBaseSites)
parameters = request.args
if parameters.get("module_code"):
query = query.where(TBaseSites.modules.any(module_code=parameters.get("module_code")))
if parameters.get("id_module"):
query = query.where(TBaseSites.modules.any(id_module=parameters.get("id_module")))
if parameters.get("id_base_site"):
query = query.where(TBaseSites.id_base_site == parameters.get("id_base_site"))
if parameters.get("base_site_name"):
query = query.where(
TBaseSites.base_site_name.ilike("%{}%".format(parameters.get("base_site_name")))
)
data = DB.session.scalars(query).all()
return [n.as_dict(fields=["id_base_site", "base_site_name"]) for n in data]
@routes.route("/siteslist/<int:id_site>", methods=["GET"])
@json_resp
[docs]
def get_onelist_site(id_site):
"""
Get minimal information for a site {id_base_site, nom site}
.. :quickref: Monitoring;
:param id_site: id of base site
:param type: int
"""
query = select(
TBaseSites.id_base_site, TBaseSites.base_site_name, TBaseSites.base_site_code
).where(TBaseSites.id_base_site == id_site)
data = DB.session.execute(query).scalar_one()
return {"id_base_site": data.id_base_site, "base_site_name": data.base_site_name}
@routes.route("/siteareas/<int:id_site>", methods=["GET"])
@json_resp
[docs]
def get_site_areas(id_site):
"""
Get areas of a site from cor_site_area as geojson
.. :quickref: Monitoring;
:param id_module: int
:type id_module: int
:param id_area_type:
:type id_area_type: int
"""
params = request.args
query = (
select(cor_site_area, func.ST_Transform(LAreas.geom, 4326))
.join(LAreas, LAreas.id_area == cor_site_area.c.id_area)
.where(cor_site_area.c.id_base_site == id_site)
)
if "id_area_type" in params:
query = query.where(LAreas.id_type == params["id_area_type"])
if "id_module" in params:
query = query.join(cor_site_module, cor_site_module.c.id_base_site == id_site).where(
cor_site_module.c.id_module == params["id_module"]
)
data = DB.session.scalars(query).all()
features = []
for d in data:
feature = get_geojson_feature(d[2])
feature["id"] = d[1]
features.append(feature)
return FeatureCollection(features)
@routes.route("/individuals/<int:id_module>", methods=["GET"])
@login_required
[docs]
def get_individuals(id_module):
action = "R"
object_code = "MONITORINGS_INDIVIDUALS"
module = DB.session.get(TModules, id_module)
if module is None:
raise NotFound("Module not found")
module_code = module.module_code
current_user = g.current_user
max_scope = get_scope(
action, id_role=current_user.id_role, module_code=module_code, object_code=object_code
)
if not max_scope:
raise Forbidden(description=_forbidden_message(action, module_code, object_code))
# FIXME: when all sqlalchemy 2.0 PR are merged, update it to fit the good practices
# like @qfilter etc...
query = select(TIndividuals).where(TIndividuals.modules.any(TModules.id_module == id_module))
results = (
DB.session.scalars(TIndividuals.filter_by_scope(query, max_scope, current_user))
.unique()
.all()
)
schema = TIndividualsSchema(exclude=["modules", "digitiser", "markings", "nomenclature_sex"])
# In the future: paginate the query. But need infinite scroll on
# select frontend side
return schema.jsonify(results, many=True)
@routes.route("/individual/<int:id_module>", methods=["POST"])
@login_required
[docs]
def create_one_individual(id_module: int):
# Id module is an optional parameter to associate an individual
# to a module
action = "C"
object_code = "MONITORINGS_INDIVIDUALS"
module = DB.session.get(TModules, id_module)
if module is None:
raise NotFound("Module not found")
module_code = module.module_code
current_user = g.current_user
max_scope = get_scope(
action, id_role=current_user.id_role, module_code=module_code, object_code=object_code
)
if not max_scope:
raise Forbidden(description=_forbidden_message(action, module_code, object_code))
# Exclude id_digitiser since it is set by the current user
individual_schema = TIndividualsSchema(exclude=["id_digitiser"], unknown=EXCLUDE)
individual_instance = TIndividuals(id_digitiser=g.current_user.id_role)
try:
individual = individual_schema.load(data=request.get_json(), instance=individual_instance)
except ValidationError as error:
raise BadRequest(error.messages)
individual.modules = [module]
DB.session.add(individual)
DB.session.commit()
return individual_schema.jsonify(individual)