chore : add SQLA 2 compatibility

.github/workflows/pytest.yml

@@ -23,7 +23,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        geonature_ref: ["master"] # Mettre version compatible
+        geonature_ref: ["develop"] # Mettre version compatible
     uses: pnx-si/geonature/.github/workflows/gn-module-pytest.yml@develop
     with:
       geonature_ref: ${{ matrix.geonature_ref }}

backend/gn_module_monitoring/blueprint.py

@@ -79,4 +79,4 @@ def set_current_module():
 for cmd in commands:
     blueprint.cli.add_command(cmd)
 
-flask_admin.add_view(BibTypeSiteView(DB.session, name="Types de site", category="Monitorings"))
+flask_admin.add_view(BibTypeSiteView(DB, name="Types de site", category="Monitorings"))

backend/gn_module_monitoring/command/imports/constant.py

@@ -85,7 +85,7 @@
     "MONITORINGS_GRP_SITES": {"label": "groupes de sites", "actions": ["C", "R", "U", "D"]},
     "MONITORINGS_SITES": {"label": "sites", "actions": ["C", "R", "U", "D"]},
     "MONITORINGS_VISITES": {"label": "visites", "actions": ["C", "R", "U", "D"]},
-    "MONITORINGS_INDIVIDUALS": {"label": "individus", "actions": ["C", "R", "U", "D"]},
+    "INDIVIDUALS": {"label": "individus", "actions": ["C", "R", "U", "D"]},
     "MONITORINGS_MARKINGS": {"label": "marquages", "actions": ["C", "R", "U", "D"]},
 }
 

backend/gn_module_monitoring/command/imports/protocol.py

@@ -294,7 +294,6 @@ def update_protocol(module_data, module_code, fields_to_delete, update_label_onl
         Booléen indiquant si la mise à jour a réussi.
     """
     try:
-        DB.session.rollback()
         module_label = module_data["module"].get("module_label")
 
         destination = DB.session.execute(
@@ -330,7 +329,7 @@ def update_protocol(module_data, module_code, fields_to_delete, update_label_onl
                 delete_bib_fields(fields_to_delete)
 
             table_name = f"t_imports_{module_code.lower()}"
-            DB.engine.execute(text(f"DROP TABLE IF EXISTS gn_imports.{table_name}"))
+            DB.session.execute(text(f"DROP TABLE IF EXISTS gn_imports.{table_name}"))
 
             create_sql_import_table_protocol(module_code, protocol_data)
 

backend/gn_module_monitoring/command/imports/sql.py

@@ -10,6 +10,7 @@
     Boolean,
     ForeignKey,
     PrimaryKeyConstraint,
+    text,
 )
 
 from geonature.utils.env import DB
@@ -35,7 +36,8 @@ def check_rows_exist_in_import_table(module_code: str) -> bool:
     table_name = f"t_imports_{module_code.lower()}"
     query = f"SELECT * FROM gn_imports.{table_name} LIMIT 1;"
     try:
-        result = DB.session.execute(query).fetchone()
+        with DB.session.begin_nested():
+            result = DB.session.execute(text(query)).fetchone()
         return result is not None
     except Exception as e:
         print(f"Erreur lors de la vérification de l'existence de la table : {str(e)}")
@@ -47,7 +49,7 @@ def create_sql_import_table_protocol(module_code: str, protocol_data):
     Create import table using SQLAlchemy metadata
     """
     table = get_imports_table_metadata(module_code, protocol_data)
-    table.metadata.create_all(DB.engine)
+    table.metadata.create_all(DB.session.connection())
     print(f"La table transitoire d'importation pour {module_code} a été créée.")
 
 

backend/gn_module_monitoring/command/sql.py

@@ -65,7 +65,7 @@ def execute_sql_file(dir, file, module_code, forbidden_instruction=[]):
         with DB.engine.begin() as conn:
             conn.execute(
                 text(sql_content),
-                module_code=module_code,
+                {"module_code": module_code},
             )
     except Exception as e:
         raise Exception("{} - erreur dans le script {} : {}".format(module_code, file, e))

backend/gn_module_monitoring/conf_schema_toml.py

@@ -14,7 +14,7 @@
     "visit": "MONITORINGS_VISITES",
     "observation": "MONITORINGS_VISITES",
     "observation_detail": "MONITORINGS_VISITES",
-    "individual": "MONITORINGS_INDIVIDUALS",
+    "individual": "INDIVIDUALS",
     "marking": "MONITORINGS_MARKINGS",
 }
 

backend/gn_module_monitoring/migrations/bbdd47114b4d_rename_individuals_permissions.py

@@ -0,0 +1,30 @@
+"""rename_individuals_permissions
+
+Revision ID: bbdd47114b4d
+Revises: 3d39820c9ab7
+Create Date: 2026-08-03 16:47:09.315441
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "bbdd47114b4d_rename_permissions"
+down_revision = "3d39820c9ab7"
+branch_labels = None
+depends_on = "ad8b797d89c0"
+
+
+def upgrade():
+    op.execute(sa.text("""
+        DELETE FROM gn_permissions.t_objects
+        WHERE code_object = 'MONITORINGS_INDIVIDUALS'
+        """))
+
+
+def downgrade():
+    op.execute(sa.text("""
+        INSERT INTO gn_permissions.t_objects (id_object, code_object, description_object)
+        VALUES(nextval('gn_permissions.t_objects_id_object_seq'::regclass), 'MONITORINGS_INDIVIDUALS', 'Permissions sur les individus');    
+        """))

backend/gn_module_monitoring/monitoring/admin.py

@@ -57,9 +57,9 @@ class BibTypeSiteView(CruvedProtectedMixin, ModelView):
     module_code = "MONITORINGS"
     object_code = "TYPES_SITES"
 
-    def __init__(self, session, **kwargs):
+    def __init__(self, db, **kwargs):
         # Référence au model utilisé
-        super(BibTypeSiteView, self).__init__(BibTypeSite, session, **kwargs)
+        super(BibTypeSiteView, self).__init__(BibTypeSite, db, **kwargs)
 
     def get_only_nomenclature_asc():
         return (

backend/gn_module_monitoring/monitoring/models.py

@@ -2,14 +2,16 @@
 Modèles SQLAlchemy pour les modules de suivi
 """
 
+from typing import Any, Optional
+
 import geoalchemy2
 
 from flask import g
 
 from uuid import uuid4
 
 from sqlalchemy import join, select, func, and_
-from sqlalchemy.orm import column_property, aliased
+from sqlalchemy.orm import Mapped, column_property, aliased, mapped_column
 
 from sqlalchemy.dialects.postgresql import JSONB, UUID
 
@@ -33,7 +35,7 @@
     corIndividualModule,
 )
 from geonature.core.gn_meta.models import TDatasets
-from geonature.core.gn_commons.models import TModules, cor_module_dataset
+from geonature.core.gn_commons.models import TModules
 from geonature.core.gn_permissions.tools import has_any_permissions_by_action
 
 from pypnusershub.db.models import User
@@ -86,11 +88,15 @@ class TMonitoringObservationDetails(DB.Model, PermissionModel, ObservationDetail
     __tablename__ = "t_observation_details"
     __table_args__ = {"schema": "gn_monitoring"}
 
-    id_observation_detail = DB.Column(DB.Integer, primary_key=True, nullable=False, unique=True)
+    id_observation_detail: Mapped[int] = mapped_column(DB.Integer, primary_key=True, unique=True)
 
-    id_observation = DB.Column(DB.ForeignKey("gn_monitoring.t_observations.id_observation"))
-    data = DB.Column(JSONB)
-    uuid_observation_detail = DB.Column(UUID(as_uuid=True), default=uuid4)
+    id_observation: Mapped[Optional[int]] = mapped_column(
+        DB.ForeignKey("gn_monitoring.t_observations.id_observation")
+    )
+    data: Mapped[Optional[Any]] = mapped_column(JSONB)
+    uuid_observation_detail: Mapped[Optional[Any]] = mapped_column(
+        UUID(as_uuid=True), default=uuid4
+    )
 
     medias = DB.relationship(
         TMedias,
@@ -117,12 +123,11 @@ class TMonitoringObservations(TObservations, PermissionModel, ObservationsQuery)
         "polymorphic_identity": "monitoring_observation",
     }
 
-    data = DB.Column(JSONB)
+    data: Mapped[Optional[Any]] = mapped_column(JSONB)
 
-    id_observation = DB.Column(
+    id_observation: Mapped[int] = mapped_column(
         DB.ForeignKey("gn_monitoring.t_observations.id_observation"),
         primary_key=True,
-        nullable=False,
     )
 
     medias = DB.relationship(
@@ -188,12 +193,12 @@ class TMonitoringVisits(TBaseVisits, PermissionModel, VisitQuery):
         "polymorphic_identity": "monitoring_visit",
     }
 
-    id_base_visit = DB.Column(
+    id_base_visit: Mapped[int] = mapped_column(
         DB.ForeignKey("gn_monitoring.t_base_visits.id_base_visit"),
         primary_key=True,
     )
 
-    data = DB.Column(JSONB)
+    data: Mapped[Optional[Any]] = mapped_column(JSONB)
 
     medias = DB.relationship(
         TMedias,
@@ -286,11 +291,11 @@ class TMonitoringSites(TBaseSites, PermissionModel, SitesQuery):
         "polymorphic_identity": "monitoring_site",
     }
 
-    id_base_site = DB.Column(
-        DB.ForeignKey("gn_monitoring.t_base_sites.id_base_site"), nullable=False, primary_key=True
+    id_base_site: Mapped[int] = mapped_column(
+        DB.ForeignKey("gn_monitoring.t_base_sites.id_base_site"), primary_key=True
     )
 
-    id_sites_group = DB.Column(
+    id_sites_group: Mapped[Optional[int]] = mapped_column(
         DB.ForeignKey(
             "gn_monitoring.t_sites_groups.id_sites_group",
         ),
@@ -300,7 +305,8 @@ class TMonitoringSites(TBaseSites, PermissionModel, SitesQuery):
         back_populates="sites",
         uselist=False,
     )
-    data = DB.Column(JSONB)
+
+    data: Mapped[Optional[Any]] = mapped_column(JSONB)
 
     modules = DB.relationship(
         "TMonitoringModules",
@@ -455,21 +461,25 @@ class TMonitoringSitesGroups(DB.Model, PermissionModel, SitesGroupsQuery):
     __tablename__ = "t_sites_groups"
     __table_args__ = {"schema": "gn_monitoring"}
 
-    id_sites_group = DB.Column(DB.Integer, primary_key=True, nullable=False, unique=True)
-    id_digitiser = DB.Column(DB.Integer, DB.ForeignKey("utilisateurs.t_roles.id_role"))
+    id_sites_group: Mapped[int] = mapped_column(DB.Integer, primary_key=True, unique=True)
+    id_digitiser: Mapped[Optional[int]] = mapped_column(
+        DB.Integer, DB.ForeignKey("utilisateurs.t_roles.id_role")
+    )
 
     digitiser = DB.relationship(
         User, primaryjoin=(User.id_role == id_digitiser), foreign_keys=[id_digitiser]
     )
-    uuid_sites_group = DB.Column(UUID(as_uuid=True), default=uuid4)
+    uuid_sites_group: Mapped[Any] = mapped_column(UUID(as_uuid=True), default=uuid4)
 
-    sites_group_name = DB.Column(DB.Unicode)
-    sites_group_code = DB.Column(DB.Unicode)
-    sites_group_description = DB.Column(DB.Unicode)
+    sites_group_name: Mapped[Optional[str]] = mapped_column(DB.Unicode)
+    sites_group_code: Mapped[Optional[str]] = mapped_column(DB.Unicode)
+    sites_group_description: Mapped[Optional[str]] = mapped_column(DB.Unicode)
 
-    comments = DB.Column(DB.Unicode)
-    geom = DB.Column(geoalchemy2.types.Geometry("GEOMETRY", 4326, nullable=True))
-    data = DB.Column(JSONB)
+    comments: Mapped[Optional[str]] = mapped_column(DB.Unicode)
+    geom: Mapped[Optional[Any]] = mapped_column(
+        geoalchemy2.types.Geometry("GEOMETRY", 4326, nullable=True)
+    )
+    data: Mapped[Optional[Any]] = mapped_column(JSONB)
 
     medias = DB.relationship(
         TMedias,
@@ -500,8 +510,8 @@ class TMonitoringSitesGroups(DB.Model, PermissionModel, SitesGroupsQuery):
         .scalar_subquery()
     )
 
-    altitude_min = DB.Column(DB.Integer)
-    altitude_max = DB.Column(DB.Integer)
+    altitude_min: Mapped[Optional[int]] = mapped_column(DB.Integer)
+    altitude_max: Mapped[Optional[int]] = mapped_column(DB.Integer)
 
     geom_geojson = column_property(
         select(func.st_asgeojson(func.st_convexHull(func.st_collect(TMonitoringSites.geom))))
@@ -573,22 +583,21 @@ class TMonitoringModules(TModules, PermissionModel, MonitoringQuery):
 
     __import_actions__ = MonitoringImportActions
 
-    id_module = DB.Column(
+    id_module: Mapped[int] = mapped_column(
         DB.ForeignKey("gn_commons.t_modules.id_module"),
         primary_key=True,
-        nullable=False,
         unique=True,
     )
 
-    uuid_module_complement = DB.Column(UUID(as_uuid=True), default=uuid4)
+    uuid_module_complement: Mapped[Any] = mapped_column(UUID(as_uuid=True), default=uuid4)
 
-    id_list_observer = DB.Column(DB.Integer)
-    id_list_taxonomy = DB.Column(DB.Integer)
-    cd_nom = DB.Column(DB.Integer)
+    id_list_observer: Mapped[Optional[int]] = mapped_column(DB.Integer)
+    id_list_taxonomy: Mapped[Optional[int]] = mapped_column(DB.Integer)
+    cd_nom: Mapped[Optional[int]] = mapped_column(DB.Integer)
 
-    taxonomy_display_field_name = DB.Column(DB.Unicode)
-    b_synthese = DB.Column(DB.Boolean)
-    b_draw_sites_group = DB.Column(DB.Boolean)
+    taxonomy_display_field_name: Mapped[Optional[str]] = mapped_column(DB.Unicode)
+    b_synthese: Mapped[Optional[bool]] = mapped_column(DB.Boolean)
+    b_draw_sites_group: Mapped[Optional[bool]] = mapped_column(DB.Boolean)
 
     medias = DB.relationship(
         TMedias,
@@ -628,17 +637,11 @@ class TMonitoringModules(TModules, PermissionModel, MonitoringQuery):
         # viewonly=True,
     )
 
-    datasets = DB.relationship(
-        "TDatasets",
-        secondary=cor_module_dataset,
-        join_depth=0,
-        overlaps="modules",
-    )
     types_site = DB.relationship(
         "BibTypeSite",
         secondary=cor_module_type,
     )
-    data = DB.Column(JSONB)
+    data: Mapped[Optional[Any]] = mapped_column(JSONB)
 
     visits = DB.relationship(
         TMonitoringVisits,

backend/gn_module_monitoring/monitoring/repositories.py

@@ -90,17 +90,19 @@ def process_synthese(self, process_module=False, limit=1000):
         table_name = "v_synthese_{}".format(self._module_code.lower())
 
         # Test de l'existance de la colonne de synchronisation sur la vue synthese
-        column_exist = DB.engine.execute(
+        column_exist = DB.session.execute(
             text("""
                 SELECT count(*)
                 FROM information_schema.columns
                 WHERE   table_schema=:table_schema
                     AND table_name=:table_name
                     AND column_name=:column_name;
                 """),
-            table_schema="gn_monitoring",
-            table_name=table_name,
-            column_name=self.config_param("id_field_name"),
+            params={
+                "table_schema": "gn_monitoring",
+                "table_name": table_name,
+                "column_name": self.config_param("id_field_name"),
+            },
         ).fetchone()
 
         if column_exist[0] == 0:

backend/gn_module_monitoring/routes/individuals.py

@@ -34,9 +34,9 @@
     methods=["GET"],
     defaults={"object_type": "individual"},
 )
-@check_cruved_scope("R", object_code="MONITORINGS_INDIVIDUALS")
+@check_cruved_scope("R", object_code="INDIVIDUALS")
 def get_individuals(object_type, module_code=None):
-    object_code = "MONITORINGS_INDIVIDUALS"
+    object_code = "INDIVIDUALS"
     params = MultiDict(request.args)
     limit, page = get_limit_page(params=params)
     sort_label, sort_dir = get_sort(
@@ -77,7 +77,7 @@ def get_individuals(object_type, module_code=None):
 @blueprint.route(
     "/individuals/<int:_id>", methods=["DELETE"], defaults={"object_type": "individual"}
 )
-@check_cruved_scope("D", get_scope=True, object_code="MONITORINGS_INDIVIDUALS")
+@check_cruved_scope("D", get_scope=True, object_code="INDIVIDUALS")
 def delete_individual(scope, _id: int, object_type: str):
     individual = db.get_or_404(TMonitoringIndividuals, _id)
     if not individual.has_instance_permission(scope=scope):

backend/gn_module_monitoring/tests/conftest.py

@@ -1,4 +1,5 @@
 import pytest
+from geonature.tests.fixtures import *
+from geonature.tests.fixtures import _session, app, _app, users
 from geonature.tests.test_monitoring import markings, nomenclature_type_markings
 from gn_module_monitoring.tests.fixtures import *
-from geonature.tests.test_users_menu import tlist

backend/gn_module_monitoring/tests/fixtures/__init__.py

@@ -1,8 +1,9 @@
 from .generic import *
 from .individual import *
-from .module import *
+from .monitoring_modules import *
 from .site import *
 from .sites_groups import *
 from .visit import *
 from .observation import *
 from .marking import *
+from .type_site import *

backend/gn_module_monitoring/tests/fixtures/individual.py

@@ -9,8 +9,6 @@
 from geonature.core.gn_monitoring.models import corIndividualModule
 from gn_module_monitoring.monitoring.models import TMonitoringIndividuals
 
-from geonature.tests.fixtures import users
-
 SPECIES = "Athene noctua"