Changement de is_habitat_complex (#3125)

* feat(occhab) : `is_habitat_complex` (bool) devient `id_nomenclature_type_mosaique_habitat` ( nomenclature)

contrib/gn_module_occhab/backend/gn_module_occhab/migrations/c1a6b0793360_replace_is_complex_with_habitat_.py

@@ -0,0 +1,190 @@
+"""replace_is_complex_with_habitat_complexity
+
+Revision ID: c1a6b0793360
+Revises: 7b6a578eccd7
+Create Date: 2024-07-18 15:52:38.695575
+
+"""
+
+from alembic import op
+from pypnnomenclature.models import TNomenclatures
+import sqlalchemy as sa
+from sqlalchemy.orm.session import Session
+
+
+# revision identifiers, used by Alembic.
+revision = "c1a6b0793360"
+down_revision = "aed662bbd88a"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+
+    conn = op.get_bind()
+    metadata = sa.MetaData(bind=conn)
+
+    # ADD new column
+    op.add_column(
+        "t_stations",
+        sa.Column(
+            "id_nomenclature_type_mosaique_habitat",
+            sa.Integer(),
+            sa.ForeignKey("ref_nomenclatures.t_nomenclatures.id_nomenclature"),
+            nullable=True,
+        ),
+        schema="pr_occhab",
+    )
+
+    # GET required nomenclature id and occhab destination id
+    session = Session(bind=op.get_bind())
+
+    id_default_type_mosaique_habitat = session.scalar(
+        sa.select(TNomenclatures.id_nomenclature).where(
+            TNomenclatures.mnemonique == "Mosaïque mixte"
+        )
+    )
+    destination = sa.Table("bib_destinations", metadata, schema="gn_imports", autoload_with=conn)
+    entity = sa.Table("bib_entities", metadata, autoload=True, schema="gn_imports")
+    theme = sa.Table("bib_themes", metadata, autoload=True, schema="gn_imports")
+    station = sa.Table("t_stations", metadata, schema="pr_occhab", autoload_with=conn)
+    cor_entity_field = sa.Table("cor_entity_field", metadata, autoload=True, schema="gn_imports")
+
+    id_theme_general = session.execute(
+        sa.select(theme.c.id_theme).where(theme.c.name_theme == "general_info")
+    ).scalar()
+    id_destination_occhab = session.scalar(
+        sa.select(destination.c.id_destination).where(destination.c.code == "occhab")
+    )
+    id_entity_station = session.execute(
+        sa.select(entity.c.id_entity).where(entity.c.code == "station")
+    ).scalar()
+
+    session.close()
+
+    # UPDATE Station table
+
+    op.execute(
+        sa.update(station)
+        .where(sa.text("pr_occhab.t_stations.is_habitat_complex = true"))
+        .values(id_nomenclature_type_mosaique_habitat=id_default_type_mosaique_habitat)
+    )
+    op.drop_column("t_stations", "is_habitat_complex", schema="pr_occhab")
+    op.execute(
+        """
+        ALTER TABLE pr_occhab.t_stations ADD CONSTRAINT 
+        check_t_stations_type_mosaique_habitat CHECK
+        (ref_nomenclatures.check_nomenclature_type_by_mnemonique(id_nomenclature_type_mosaique_habitat, 'MOSAIQUE_HAB'::character varying)) NOT VALID
+        """
+    )
+    ## UPDATE Transient table
+    op.add_column(
+        "t_imports_occhab",
+        sa.Column("src_id_nomenclature_type_mosaique_habitat", sa.Unicode, nullable=True),
+        schema="gn_imports",
+    )
+    op.add_column(
+        "t_imports_occhab",
+        sa.Column(
+            "id_nomenclature_type_mosaique_habitat",
+            sa.Integer,
+            sa.ForeignKey("ref_nomenclatures.t_nomenclatures.id_nomenclature"),
+        ),
+        schema="gn_imports",
+    )
+    op.drop_column("t_imports_occhab", "src_is_habitat_complex", schema="gn_imports")
+    op.drop_column("t_imports_occhab", "is_habitat_complex", schema="gn_imports")
+
+    # UPDATE BibFields
+    bib_fields = sa.Table("bib_fields", metadata, schema="gn_imports", autoload_with=conn)
+    op.execute(sa.delete(bib_fields).where(bib_fields.c.name_field == "is_habitat_complex"))
+    id_new_field = (
+        op.get_bind()
+        .execute(
+            sa.insert(bib_fields)
+            .values(
+                name_field="id_nomenclature_type_mosaique_habitat",
+                fr_label="Mosaïque d'habitats",
+                source_field="src_id_nomenclature_type_mosaique_habitat",
+                dest_field="id_nomenclature_type_mosaique_habitat",
+                mandatory=False,
+                mnemonique="MOSAIQUE_HAB",
+                autogenerated=False,
+                display=True,
+                id_destination=id_destination_occhab,
+            )
+            .returning(bib_fields.c.id_field)
+        )
+        .scalar()
+    )
+
+    op.execute(
+        sa.insert(cor_entity_field).values(
+            id_entity=id_entity_station,
+            id_field=id_new_field,
+            id_theme=id_theme_general,
+            comment="Correspondance champs standard: mosaiqueValue",
+            order_field=op.get_bind()
+            .execute(
+                sa.select(sa.func.max(cor_entity_field.c.order_field) + 1).select_from(
+                    cor_entity_field
+                )
+            )
+            .scalar(),
+        )
+    )
+
+
+def downgrade():
+    conn = op.get_bind()
+    metadata = sa.MetaData(bind=conn)
+
+    session = Session(bind=op.get_bind())
+
+    Destination = sa.Table("bib_destinations", metadata, schema="gn_imports", autoload_with=conn)
+    id_destination_occhab = session.scalar(
+        sa.select(Destination.c.id_destination).where(Destination.c.code == "occhab")
+    )
+    session.close()
+
+    op.drop_constraint("check_t_stations_type_mosaique_habitat", "t_stations", schema="pr_occhab")
+    op.drop_column("t_stations", "id_nomenclature_type_mosaique_habitat", schema="pr_occhab")
+    op.add_column(
+        "t_stations",
+        sa.Column("is_habitat_complex", sa.Boolean, nullable=True),
+        schema="pr_occhab",
+    )
+    op.drop_column("t_imports_occhab", "id_nomenclature_type_mosaique_habitat", schema="gn_imports")
+    op.drop_column(
+        "t_imports_occhab", "src_id_nomenclature_type_mosaique_habitat", schema="gn_imports"
+    )
+    op.add_column(
+        "t_imports_occhab",
+        sa.Column("src_is_habitat_complex", sa.Unicode, nullable=True),
+        schema="gn_imports",
+    )
+    op.add_column(
+        "t_imports_occhab",
+        sa.Column("is_habitat_complex", sa.Boolean, nullable=True),
+        schema="gn_imports",
+    )
+    BibFields = sa.Table(
+        "bib_fields", sa.MetaData(), schema="gn_imports", autoload_with=op.get_bind()
+    )
+    op.execute(
+        sa.delete(BibFields).where(
+            BibFields.c.name_field == "id_nomenclature_type_mosaique_habitat"
+        )
+    )
+    op.execute(
+        sa.insert(BibFields).values(
+            name_field="is_habitat_complex",
+            fr_label="Complexité d'habitat",
+            source_field="src_is_habitat_complex",
+            dest_field="is_habitat_complex",
+            mandatory=False,
+            autogenerated=False,
+            display=True,
+            id_destination=id_destination_occhab,
+        )
+    )

contrib/gn_module_occhab/backend/gn_module_occhab/models.py

@@ -48,7 +48,15 @@ class Station(NomenclaturesMixin, db.Model):
     date_max = db.Column(db.DateTime, server_default=FetchedValue())
     observers_txt = db.Column(db.Unicode(length=500))
     station_name = db.Column(db.Unicode(length=1000))
-    is_habitat_complex = db.Column(db.Boolean)
+    # is_habitat_complex = db.Column(db.Boolean)
+    id_nomenclature_type_mosaique_habitat = db.Column(
+        db.Integer,
+        ForeignKey(Nomenclature.id_nomenclature),
+    )
+    type_mosaique_habitat = db.relationship(
+        Nomenclature,
+        foreign_keys=[id_nomenclature_type_mosaique_habitat],
+    )
     altitude_min = db.Column(db.Integer)
     altitude_max = db.Column(db.Integer)
     depth_min = db.Column(db.Integer)