Use upsert strategy

- Add field admindiv.gaul
- Do not use TRUNCATE CASCADE
- Use UPSERT strategy for admindivs
- Maintain contacts and recommandations relations using GAUL
- Add class docstring

Author: arnaud-morvan

alembic/versions/50e54a034fa5_add_field_admindiv_gaul.py

@@ -0,0 +1,39 @@
+"""Add field admindiv.gaul
+
+Revision ID: 50e54a034fa5
+Revises: cdcea4ae97ed
+Create Date: 2025-12-19 12:21:52.532645
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '50e54a034fa5'
+down_revision = 'cdcea4ae97ed'
+branch_labels = None
+depends_on = None
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade(engine_name):
+    op.add_column(
+        'administrativedivision',
+        sa.Column('gaul', sa.Integer(), nullable=True),
+        schema='datamart',
+    )
+    op.create_index(
+        op.f('ix_datamart_administrativedivision_gaul'),
+        'administrativedivision',
+        ['gaul'],
+        # unique=True,
+        schema='datamart',
+    )
+
+def downgrade(engine_name):
+    op.drop_index(
+        op.f('ix_datamart_administrativedivision_gaul'),
+        table_name='administrativedivision',
+        schema='datamart',
+    )
+    op.drop_column('administrativedivision', 'gaul', schema='datamart')

tests/data/ADM2.geojson

@@ -1,6 +1,6 @@
 {
   "type": "FeatureCollection",
-  "name": "TH_ADM2_sample",
+  "name": "ADM2",
   "crs": {
     "type": "name",
     "properties": {

tests/data/URB.geojson

@@ -1,6 +1,6 @@
 {
   "type": "FeatureCollection",
-  "name": "TH_URB_sample",
+  "name": "URB",
   "crs": {
     "type": "name",
     "properties": {

tests/processing/test_import_geopackage.py

@@ -22,7 +22,14 @@
 import geopandas as gpd
 from mock import patch
 
-from thinkhazard.models import AdministrativeDivision
+from thinkhazard.models import (
+    AdministrativeDivision,
+    ClimateChangeRecommendation,
+    ClimateChangeRecAdministrativeDivisionAssociation,
+    Contact,
+    ContactAdministrativeDivisionHazardTypeAssociation,
+    HazardType,
+)
 from thinkhazard.processing.import_geopackage import GeopackageImporter
 
 from .. import DBSession, engine, settings, DATA_FOLDER
@@ -39,9 +46,9 @@
 
 def mock_read_file(path, layer=None, **kwargs):
     if layer == "ADM2":
-        return _real_read_file(ADM2_PATH, engine="pyogrio")
+        return _real_read_file(ADM2_PATH, layer=layer, engine="pyogrio")
     if layer == "URB":
-        return _real_read_file(URBAN_PATH, engine="pyogrio")
+        return _real_read_file(URBAN_PATH, layer=layer, engine="pyogrio")
     else:
         raise ValueError(f"Unknown layer: {layer}")
 
@@ -76,3 +83,72 @@ def test_dry_run(self, mock_read):
         assert DBSession.query(AdministrativeDivision).count() == 3
         self.importer().execute(geopackage_path=ADM2_PATH, verbose=True, dry_run=True)
         assert DBSession.query(AdministrativeDivision).count() == 3
+
+    @patch(
+        "thinkhazard.processing.import_geopackage.gpd.read_file",
+        side_effect=mock_read_file,
+    )
+    def test_contact_gaul(self, mock_read):
+        """Test that contacts are correctly mapped using GAUL."""
+
+        administrativedivision = AdministrativeDivision(
+            code="999999",  # GAUL
+            leveltype_id=1,
+            name="test",
+        )
+        DBSession.add(administrativedivision)
+
+        contact = Contact(
+            name="test",
+        )
+        DBSession.add(contact)
+
+        hazardtype = DBSession.query(HazardType).first()
+
+        rel = ContactAdministrativeDivisionHazardTypeAssociation(
+            contact=contact,
+            administrativedivision=administrativedivision,
+            hazardtype=hazardtype,
+        )
+        DBSession.add(rel)
+        DBSession.flush()
+
+        self.importer().execute(geopackage_path=ADM2_PATH, verbose=True)
+
+        DBSession.expire_all()
+        assert rel.administrativedivision.code == "TST"
+
+    @patch(
+        "thinkhazard.processing.import_geopackage.gpd.read_file",
+        side_effect=mock_read_file,
+    )
+    def test_climatechangerecommendation_gaul(self, mock_read):
+        """Test that climatechangerecommandations are correctly mapped using GAUL."""
+
+        administrativedivision = AdministrativeDivision(
+            code="999999",  # GAUL
+            leveltype_id=1,
+            name="test",
+        )
+        DBSession.add(administrativedivision)
+
+        hazardtype = DBSession.query(HazardType).first()
+
+        climatechangerecommendation = ClimateChangeRecommendation(
+            text="test",
+            hazardtype=hazardtype,
+        )
+        DBSession.add(climatechangerecommendation)
+
+        rel = ClimateChangeRecAdministrativeDivisionAssociation(
+            climatechangerecommendation=climatechangerecommendation,
+            administrativedivision=administrativedivision,
+            hazardtype=hazardtype,
+        )
+        DBSession.add(rel)
+        DBSession.flush()
+
+        self.importer().execute(geopackage_path=ADM2_PATH, verbose=True)
+
+        DBSession.expire_all()
+        assert rel.administrativedivision.code == "TST"

thinkhazard/models.py

@@ -282,6 +282,7 @@ class AdministrativeDivision(Base):
 
     id = Column(Integer, primary_key=True)
     code = Column(String, index=True, unique=True, nullable=False)
+    gaul = Column(Integer, index=True)
     leveltype_id = Column(
         Integer, ForeignKey(AdminLevelType.id), nullable=False, index=True
     )