Habref update by Pierre-Narcisi · Pull Request #27 · PnX-SI/Habref-api-module

Pierre-Narcisi · 2026-07-02T11:40:24Z

Ajout de commandes pour mettre à jour habref

jacquesfize · 2026-07-02T11:50:05Z

    corresp_syn = DB.Column(DB.Boolean)


+@serializable


Peut être plus future proof de passer par les schémas marshmallow. Notamment, avec l'utilisation de https://github.com/PnX-SI/Utils-Flask-SQLAlchemy/blob/master/src/utils_flask_sqla/schema.py

jacquesfize · 2026-07-02T11:56:28Z

+## Commandes disponibles
+
+Les commandes sont accessibles depuis le virtualenv de GeoNature :
+
+```bash
+cd ~/geonature
+source backend/venv/bin/activate
+```
+
+---


Suggested change

## Commandes disponibles

Les commandes sont accessibles depuis le virtualenv de GeoNature :

```bash

cd ~/geonature

source backend/venv/bin/activate

```

---

jacquesfize · 2026-07-02T12:05:58Z

+---
+
+## Étape 1 — Importer la nouvelle version et détecter les orphelins
+


Pour moi, je serais plus succinct .
"""

Dans la première étape, il faut télécharger les données du référentiels, stocker ces dernières dans une table temporaire (ref_habitats.tmp_<num_version>). Pour cela, on lance la commande :

geonature habref import-v07

[!DANGER]
Il se peut que certaines entrées du référentiels soient supprimées lors d'une mise à jours. Dans ce cas, le fichier tmp/habref/orphans_habref.csv liste l'ensemble des données dans votre base utilisant ces entrées Habref.

Le CSV généré contient les colonnes suivantes :

Colonne Description

ref_table Table du référentiel concernée (ex: habref, typoref)

table_name Table applicative contenant la référence orpheline

schema Schéma de cette table

fk_column Colonne portant la clé étrangère

fk_value Valeur orpheline

nb_lignes_affectees Nombre de lignes concernées

"""

jacquesfize · 2026-07-02T12:13:04Z

+
+## Étape 3 — Appliquer la mise à jour
+
+Une fois les données orphelines corrigées :


Suggested change

Une fois les données orphelines corrigées :

Une fois les données orphelines corrigées, lancez la mise à jour effective du référentiel à l'aide de la commande suivante :

jacquesfize · 2026-07-02T12:14:01Z

+5. Réactive les contraintes de clés étrangères
+6. Supprime les tables temporaires `tmp_*`
+
+> ⚠️ Si des données orphelines subsistent au moment du `apply-v07`, des erreurs de contraintes FK pourraient apparaître et la migration ne se feras pas. Assurez-vous que toutes les corrections ont bien été appliquées avant de lancer cette commande.


Suggested change

> ⚠️ Si des données orphelines subsistent au moment du `apply-v07`, des erreurs de contraintes FK pourraient apparaître et la migration ne se feras pas. Assurez-vous que toutes les corrections ont bien été appliquées avant de lancer cette commande.

> [!WARNING]

> ⚠️ Si des données orphelines subsistent au moment du `apply-v07`, des erreurs de contraintes FK pourraient apparaître et la migration ne se feras pas. Assurez-vous que toutes les corrections ont bien été appliquées avant de lancer cette commande.

jacquesfize · 2026-07-02T13:23:36Z

+}
+
+
+def import_habref(logger, num_version, habref_archive_name):


à mettre dans un fichier séparé vu que c'est générique

jacquesfize · 2026-07-02T13:30:12Z

+def copy_from_csv(
+    f,
+    table_name,
+    table_fields,
+    schema="",
+    header=True,
+    encoding=None,
+    delimiter=None,
+    db=None,
+):
+    bind = db.session.get_bind()
+    metadata = MetaData(bind=bind)
+    engine = db.engine
+
+    table_fields_list = list(table_fields.values())
+    table_fields_key_list = list(table_fields.keys())
+
+    final_table_name = table_name
+    table_name = f"import_{table_name}"
+    field_names = get_csv_field_names(f, encoding=encoding, delimiter=delimiter)
+    field_names = list(map(lambda field_name: field_name.replace("\ufeff", ""), field_names))
+    table = Table(
+        table_name,
+        metadata,
+        *[sa.Column(c, sa.String) for c in map(str.lower, field_names)],
+        schema=schema,
+    )
+    table.create(bind=db.session.connection())
+
+    options = ["FORMAT CSV"]
+    if header:
+        options.append("HEADER")
+    if encoding:
+        options.append(f"ENCODING '{encoding}'")
+    if delimiter:
+        options.append(f"DELIMITER E'{delimiter}'")
+    options = ", ".join(options)
+    cursor = db.session.connection().connection.cursor()
+    cursor.copy_expert(
+        f"""
+        COPY {schema}.{table_name}
+        FROM STDIN WITH ({options})
+    """,
+        f,
+    )
+
+    testTable = Table(
+        final_table_name, MetaData(), schema=schema, autoload_with=db.session.connection()
+    )
+
+    for col in testTable.columns:
+        if col.name in table_fields:
+            table_fields[col.name] = f"{table_fields[col.name]}::{col.type}"
+    table_fields_list = list(table_fields.values())
+
+    db.session.execute(
+        f"""
+    INSERT INTO {schema}.{final_table_name} ({", ".join(table_fields_key_list)})
+        SELECT {", ".join(table_fields_list)}
+        FROM {schema}.{table_name};
+    """
+    )
+    table.drop(bind=db.session.connection())


à mettre dans utils flask sqlalchemy, car une même fonction faite dans TaxHub https://github.com/PnX-SI/TaxHub/blob/1158c0c1d104367d827dfb33c704baa6a98f958a/apptax/migrations/utils.py#L26

jacquesfize · 2026-07-02T13:31:57Z

+def empty_table(table_name, db, schema=""):
+    inspector = sa_inspect(db.engine)
+
+    # Récupère les foreign keys
+    foreign_keys = inspector.get_foreign_keys(table_name, schema=schema)
+    db.session.execute("SET session_replication_role = 'replica'")
+    for fk in foreign_keys:
+        constraint_name = fk["name"]
+        if constraint_name:  # Vérifie que le nom existe
+            db.session.execute(
+                f"ALTER TABLE {schema}.{table_name} DROP CONSTRAINT {constraint_name} CASCADE"
+            )
+    db.session.execute(f"DELETE FROM {schema}.{table_name}")
+    return foreign_keys
+
+
+def restore_constraints(table_name, db, constraints, schema=""):
+    if constraints is None:
+        return
+    for constraint in constraints:
+        if isinstance(constraint, PrimaryKeyConstraint):
+            cols = ", ".join(str(col).split(".")[1] for col in constraint.columns)
+            db.session.execute(
+                f"ALTER TABLE {schema}.{table_name} ADD CONSTRAINT {constraint.name} UNIQUE ({cols})"
+            )
+        elif isinstance(constraint, ForeignKeyConstraint):
+            cols = ", ".join(str(col).split(".")[1] for col in constraint.columns)
+            ref_cols = ", ".join(
+                str(col).split(".")[1] for col in constraint.references[0].columns
+            )
+            db.session.execute(
+                f"ALTER TABLE {schema}.{table_name} ADD CONSTRAINT {constraint.name} FOREIGN KEY {cols} REFERENCES {constraint.references[0].table.name}({ref_cols})"
+            )
+
+
+def get_referencing_tables(table_name, db, schema="", exclude_tables=None):
+    """Trouve toutes les tables qui ont des FK pointant vers table_name,
+    en excluant les tables listées dans exclude_tables."""
+    exclude_tables = set(exclude_tables or [])
+    inspector = sa_inspect(db.engine)
+    referencing_tables = []
+
+    for other_schema in inspector.get_schema_names():
+        if other_schema in ("pg_catalog", "information_schema", "pg_toast"):
+            continue
+        try:
+            for other_table in inspector.get_table_names(schema=other_schema):
+                if other_schema == schema and other_table in exclude_tables:
+                    continue
+                for fk in inspector.get_foreign_keys(other_table, schema=other_schema):
+                    referred_schema = fk.get("referred_schema") or other_schema
+                    if fk["referred_table"] == table_name and (
+                        not schema or referred_schema == schema
+                    ):
+                        referencing_tables.append(
+                            {
+                                "schema": other_schema,
+                                "table": other_table,
+                                "fk_column": fk["constrained_columns"][0],
+                                "ref_column": fk["referred_columns"][0],
+                            }
+                        )
+        except Exception as e:
+            print(f"Impossible d'inspecter le schéma {other_schema}: {e}")
+
+    return referencing_tables
+
+
+CSV_FIELDNAMES = [
+    "ref_table",
+    "table_name",
+    "schema",
+    "fk_column",
+    "fk_value",
+    "nb_lignes_affectees",
+]
+
+
+def collect_orphan_rows(ref_table, new_ref_table, pk_col, db, schema="", exclude_tables=None):
+    """
+    Retourne la liste des lignes orphelines pour une table du référentiel :
+    valeurs de pk_col présentes dans les tables référençantes mais absentes de new_ref_table.
+    """
+    referencing = get_referencing_tables(
+        ref_table, db, schema=schema, exclude_tables=exclude_tables
+    )
+    rows = []
+    for ref in referencing:
+        fk_col = ref["fk_column"]
+        src = sa_table(ref["table"], sa_column(fk_col), schema=ref["schema"])
+        ref_t = sa_table(new_ref_table, sa_column(pk_col), schema=schema)
+        subq = select(1).select_from(ref_t).where(ref_t.c[pk_col] == src.c[fk_col])
+        stmt = (
+            select(src.c[fk_col], func.count().label("nb_lignes"))
+            .where(src.c[fk_col].isnot(None))
+            .where(~exists(subq))
+            .group_by(src.c[fk_col])
+            .order_by(src.c[fk_col])
+        )
+        for fk_value, nb_lignes in db.session.execute(stmt).fetchall():
+            rows.append(
+                {
+                    "ref_table": ref_table,
+                    "table_name": ref["table"],
+                    "schema": ref["schema"],
+                    "fk_column": fk_col,
+                    "fk_value": fk_value,
+                    "nb_lignes_affectees": nb_lignes,
+                }
+            )
+    return rows
+
+
+def export_orphans_to_csv(rows, output_path):
+    """Écrit la liste de lignes orphelines dans un CSV."""
+    if not rows:
+        return 0
+    dirname = os.path.dirname(output_path)
+    if dirname:
+        os.makedirs(dirname, exist_ok=True)
+    with open(output_path, "w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=CSV_FIELDNAMES)
+        writer.writeheader()
+        writer.writerows(rows)
+    return len(rows)


même remarque, je pense que c'est des fonctions qui mériteraient d'être générique

Pierre-Narcisi added 9 commits July 2, 2026 13:50

Update habref

68d4b64

change constrainte handling

c0efab4

wip

39d6800

merge

d4b4b82

wip

e713e26

Finish habref update

75e9ae5

extend to other tables than habref

0897cf9

add doc for migration

77b3691

add test and apply black

2d359fa

Pierre-Narcisi force-pushed the habref-update branch from 2ca6394 to 2d359fa Compare July 2, 2026 11:54

add command to pyproj

7876969

jacquesfize reviewed Jul 2, 2026

View reviewed changes

Pierre-Narcisi added 2 commits July 16, 2026 16:07

apply fixes

1ad1f2c

bump utils-flask-sqlalchemy

0fe1ca9

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Habref update - #27

Habref update#27
Pierre-Narcisi wants to merge 12 commits into
developfrom
habref-update

Pierre-Narcisi commented Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

Uh oh!

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

jacquesfize Jul 2, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		---

		## Étape 1 — Importer la nouvelle version et détecter les orphelins

Colonne	Description
`ref_table`	Table du référentiel concernée (ex: `habref`, `typoref`)
`table_name`	Table applicative contenant la référence orpheline
`schema`	Schéma de cette table
`fk_column`	Colonne portant la clé étrangère
`fk_value`	Valeur orpheline
`nb_lignes_affectees`	Nombre de lignes concernées


		## Étape 3 — Appliquer la mise à jour

		Une fois les données orphelines corrigées :

	> ⚠️ Si des données orphelines subsistent au moment du `apply-v07`, des erreurs de contraintes FK pourraient apparaître et la migration ne se feras pas. Assurez-vous que toutes les corrections ont bien été appliquées avant de lancer cette commande.
	> [!WARNING]
	> ⚠️ Si des données orphelines subsistent au moment du `apply-v07`, des erreurs de contraintes FK pourraient apparaître et la migration ne se feras pas. Assurez-vous que toutes les corrections ont bien été appliquées avant de lancer cette commande.

		}


		def import_habref(logger, num_version, habref_archive_name):

Uh oh!

Conversation

Pierre-Narcisi commented Jul 2, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants