Code source de geonature.core.gn_synthese.utils.blurring

from functools import lru_cache

from flask import g
import sqlalchemy as sa
from sqlalchemy.orm import aliased
from pypnnomenclature.models import BibNomenclaturesTypes, TNomenclatures
from ref_geo.models import BibAreasTypes, LAreas

from geonature.core.gn_synthese.models import CorAreaSynthese, Synthese, VSyntheseForWebApp
from geonature.core.sensitivity.models import cor_sensitivity_area_type
from geonature.core.gn_synthese.utils.query_select_sqla import SyntheseQuery

[docs] def split_blurring_precise_permissions(permissions): """ Return permissions respectively with and without sensitivity filter. """ return [p for p in permissions if p.sensitivity_filter], [ p for p in permissions if not p.sensitivity_filter ]
@lru_cache() # to retrive non sensitive nomenclature only on first call
[docs] def build_sensitive_unsensitive_filters(): """ Return where clauses for sensitive and non-sensitive observations. """ non_sensitive_nomenc = ( TNomenclatures.query.with_entities(TNomenclatures.id_nomenclature) .filter( TNomenclatures.nomenclature_type.has(BibNomenclaturesTypes.mnemonique == "SENSIBILITE") ) .filter(TNomenclatures.cd_nomenclature == "0") .one() ) return ( Synthese.id_nomenclature_sensitivity != non_sensitive_nomenc.id_nomenclature, Synthese.id_nomenclature_sensitivity == non_sensitive_nomenc.id_nomenclature, )
[docs] def build_blurred_precise_geom_queries( filters, where_clauses: list = [], select_size_hierarchy=False ): # Build 2 queries that will be UNIONed # The where_clauses list enables to add more conditions to the base query # Used in export query where_clauses.append(Synthese.the_geom_4326.isnot(None)) # Query precise geom, for use with unsensitive observations # and sensitive observations with precise permission columns = [ sa.literal(1).label("priority"), Synthese.id_synthese.label("id_synthese"), Synthese.the_geom_4326.label("geom"), ] # Size hierarchy can be used here to filter on it in # a mesh mode scenario. if select_size_hierarchy: # 0 since no blurring geometry is associated here and a point have a 0 size columns.append(sa.literal(0).label("size_hierarchy")) precise_geom_query = SyntheseQuery( Synthese,*columns).where(sa.and_(*where_clauses)).order_by(Synthese.id_synthese.desc()), filters=dict(filters), # not to edit the actual filter object ) # In both queries, we applied all filters so that we do not need to query the # whole synthese table precise_geom_query.filter_taxonomy() precise_geom_query.filter_other_filters(g.current_user) precise_geom_query.build_query() # Query blurred geom, for use with sensitive observations CorAreaSyntheseAlias = aliased(CorAreaSynthese) LAreasAlias = aliased(LAreas) BibAreasTypesAlias = aliased(BibAreasTypes) geom = LAreasAlias.geom.st_transform(4326).label("geom") # In SyntheseQuery below : # - query_joins parameter is needed to bypass # "self.query_joins is not None" condition in the build_query() method below # - priority is used to prevail non blurred geom over blurred geom if the user # can access to the non blurred geom # - orderby needed to match the non blurred and the blurred observations columns = [ sa.literal(2).label("priority"), Synthese.id_synthese.label("id_synthese"), geom, ] # size hierarchy is the size of the joined blurring area if select_size_hierarchy: columns.append(BibAreasTypes.size_hierarchy.label("size_hierarchy")) blurred_geom_query = SyntheseQuery( Synthese,*columns) .where( cor_sensitivity_area_type.c.id_nomenclature_sensitivity == Synthese.id_nomenclature_sensitivity ) .where(sa.and_(*where_clauses)) .order_by(Synthese.id_synthese.desc()), filters=dict(filters), query_joins=sa.join( Synthese, CorAreaSyntheseAlias, CorAreaSyntheseAlias.id_synthese == Synthese.id_synthese, ), geom_column=LAreas.geom_4326, ) # Joins here are needed to retrieve the blurred geometry blurred_geom_query.add_join(LAreasAlias, LAreasAlias.id_area, CorAreaSyntheseAlias.id_area) blurred_geom_query.add_join(BibAreasTypesAlias, BibAreasTypesAlias.id_type, LAreasAlias.id_type) blurred_geom_query.add_join( cor_sensitivity_area_type, cor_sensitivity_area_type.c.id_area_type, BibAreasTypesAlias.id_type, ) # Same for the first query => apply filter to avoid querying the whole table blurred_geom_query.filter_taxonomy() blurred_geom_query.filter_other_filters(g.current_user) blurred_geom_query.build_query() return blurred_geom_query, precise_geom_query
[docs] def build_allowed_geom_cte( blurring_permissions, precise_permissions, blurred_geom_query, precise_geom_query, limit, ): """ """ # The goal is to separate the blurring and precise permissions. # But in sensitive permissions there can be unsensitive observations so we need # to split them. # sensitive_where_clause and unsensitive_where_clause represents this split # See sensitive_obs_filter, unsensitive_obs_filter = build_sensitive_unsensitive_filters() # Note: the used query is not important here, as it is only used to select the right Synthese model precise_perms_filter = precise_geom_query.build_permissions_filter( g.current_user, precise_permissions, ) blurring_perms_filter = precise_geom_query.build_permissions_filter( g.current_user, blurring_permissions, ) # Access precise geom for obs with precise perm and for unsensitive obs with blurring perm precise_geom_query = precise_geom_query.query.where( sa.or_( precise_perms_filter, sa.and_( blurring_perms_filter, unsensitive_obs_filter, ), ) ).limit(limit) # Access blurred geom for sensitive obs with blurring perms blurred_geom_query = blurred_geom_query.query.where( sa.and_( blurring_perms_filter, sensitive_obs_filter, ) ).limit(limit) return precise_geom_query.union(blurred_geom_query).cte("allowed_geom")
[docs] def build_synthese_obs_query(observations, allowed_geom_cte, limit): # Final observation query # orderby priority as explained in build_allowed_geom_cte() obs_query = ( .select_from( VSyntheseForWebApp.__table__.join( allowed_geom_cte, allowed_geom_cte.c.id_synthese == VSyntheseForWebApp.id_synthese ) ) .order_by(VSyntheseForWebApp.id_synthese, allowed_geom_cte.c.priority) .distinct(VSyntheseForWebApp.id_synthese) .limit(limit) ) return obs_query