add referencial managing functions

pyproject.toml

@@ -38,6 +38,7 @@ tests = [
     "flask-marshmallow",
     "marshmallow-sqlalchemy",
     "psycopg2-binary",
+    "pytest-cov"
 ]
 
 [project.urls]

src/utils_flask_sqla/referential.py

@@ -0,0 +1,99 @@
+import csv
+import os
+
+from sqlalchemy import (
+    inspect as sa_inspect,
+    func,
+    exists,
+    select,
+    table as sa_table,
+    column as sa_column,
+)
+
+CSV_FIELDNAMES = [
+    "ref_table",
+    "table_name",
+    "schema",
+    "fk_column",
+    "fk_value",
+    "nb_affected_lines",
+]
+
+
+def get_referencing_tables(table_name, db, schema="", exclude_tables=[]):
+    """Trouve toutes les tables qui ont des FK pointant vers table_name,
+    en excluant les tables listées dans exclude_tables (même schéma)."""
+    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_table != table_name and other_table in exclude_tables:
+                    continue
+                for fk in inspector.get_foreign_keys(other_table, schema=other_schema):
+                    if fk["referred_table"] == f"{schema}.{table_name}":
+                        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
+
+
+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_lines"))
+            .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_lines 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_affected_lines": nb_lines,
+                }
+            )
+    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)

src/utils_flask_sqla/tests/psql_tests/conftest.py

@@ -0,0 +1,88 @@
+import importlib.util
+import os
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import OperationalError
+
+from utils_flask_sqla.revision import alter_table_with_dependent_views
+
+# ---------------------------------------------------------------------------
+# Integration tests against a real PostgreSQL database.
+#
+# These exercise the actual `pg_capture_dependent_views` / `pg_drop_dependent_views`
+# / `pg_recreate_dependent_views` SQL functions (created by migration revision
+# 1d09a9b67970) against a real dependency chain: a table, a materialized view
+# built on that table, and a plain view built on that materialized view.
+#
+# They require a running PostgreSQL server. Point TEST_PG_DATABASE_URI at it,
+# e.g.:
+#   docker run --rm -d -p 5432:5432 -e POSTGRES_USER=geonatadmin \
+#       -e POSTGRES_PASSWORD=geonatadmin -e POSTGRES_DB=geonature2db \
+#       postgis/postgis:15-3.4
+# Tests are skipped automatically if no server is reachable.
+# ---------------------------------------------------------------------------
+
+PG_URI = os.environ.get(
+    "TEST_PG_DATABASE_URI",
+    "postgresql+psycopg2://geonatadmin:geonatadmin@localhost:5432/geonature2db",
+)
+
+MIGRATION_FILE = (
+    Path(__file__).resolve().parent.parent.parent
+    / "migrations"
+    / "versions"
+    / "1d09a9b67970_add_functions_to_fetch_dependent_view_.py"
+)
+
+
+def _load_migration_module():
+    spec = importlib.util.spec_from_file_location(
+        "utils_flask_sqla_test_dependent_views_migration", MIGRATION_FILE
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture(scope="module")
+def pg_engine():
+    from alembic.operations import Operations
+    from alembic.runtime.migration import MigrationContext
+
+    engine = create_engine(PG_URI)
+    try:
+        with engine.connect() as conn:
+            conn.execute(text("SELECT 1"))
+    except OperationalError as exc:
+        engine.dispose()
+        pytest.skip(f"PostgreSQL test database not reachable at {PG_URI}: {exc}")
+
+    migration = _load_migration_module()
+    with engine.begin() as conn:
+        with Operations.context(MigrationContext.configure(conn)):
+            migration.upgrade()
+
+    yield engine
+
+    with engine.begin() as conn:
+        with Operations.context(MigrationContext.configure(conn)):
+            migration.downgrade()
+    engine.dispose()
+
+
+@pytest.fixture
+def pg_conn(pg_engine):
+    """A connection wrapping the whole test in one rolled-back transaction.
+
+    PostgreSQL DDL is transactional, so creating the schema/table/views inside
+    this transaction and rolling it back at teardown leaves no trace, without
+    needing per-test unique names.
+    """
+    with pg_engine.connect() as conn:
+        trans = conn.begin()
+        try:
+            yield conn
+        finally:
+            trans.rollback()

src/utils_flask_sqla/tests/psql_tests/test_referential.py

@@ -0,0 +1,112 @@
+import pytest
+
+
+from tempfile import NamedTemporaryFile
+
+import pytest
+from flask import Flask
+from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy import ForeignKey, Integer, Table, Unicode, insert
+from utils_flask_sqla.referential import collect_orphan_rows, get_referencing_tables
+from utils_flask_sqla.tests.psql_tests.conftest import PG_URI
+from utils_flask_sqla.tests.utils import TestSession
+
+db = SQLAlchemy()
+
+
+table_1 = Table(
+    "public.table1",
+    db.metadata,
+    db.Column("pk", Integer, primary_key=True),
+    db.Column("name", Unicode),
+)
+table_2 = Table(
+    "public.table2",
+    db.metadata,
+    db.Column("pk", Integer, primary_key=True),
+    db.Column("fk", Integer, ForeignKey(table_1.c.pk)),
+)
+table_1_new = Table(
+    "public.table1_new",
+    db.metadata,
+    db.Column("pk", Integer, primary_key=True),
+)
+
+
+@pytest.fixture(scope="session")
+def _app():
+    app = Flask(__name__)
+    app.config["SQLALCHEMY_DATABASE_URI"] = PG_URI
+    db.init_app(app)
+    with app.app_context():
+        db.create_all()
+        yield
+
+
+@pytest.fixture(scope="session")
+def _session(_app):
+    db.session.session_factory.class_ = TestSession
+    db.session.remove()
+    return db.session
+
+
+@pytest.fixture(scope="session")
+def app(_app, _session):
+    pass
+
+
+@pytest.fixture(scope="session")
+def data(app):
+    with db.session.begin_nested():
+        db.session.execute(
+            insert(table_1).values(
+                [
+                    {"pk": 1, "name": "ligne1"},
+                    {"pk": 2, "name": "ligne2"},
+                    {"pk": 3, "name": "ligne3"},
+                ]
+            )
+        )
+        db.session.execute(
+            insert(table_2).values(
+                [
+                    {"pk": 1, "fk": 1},
+                    {"pk": 2, "fk": 2},
+                    {"pk": 3, "fk": 3},
+                ]
+            )
+        )
+
+
+@pytest.mark.usefixtures("app")
+class TestReferential:
+
+    def test_get_get_referencing_tables(data):
+        referenced_tables = get_referencing_tables("table1", db, "public")
+        assert any([table_def["table"] == "public.table2" for table_def in referenced_tables])
+
+    def test_collect_orphan_rows_no_orphans(self, data):
+        rows = collect_orphan_rows("table1", "public.table1", "pk", db, schema="public")
+        assert rows == []
+
+    def test_collect_orphan_rows_detects_missing_rows(self, data):
+        db.session.execute(insert(table_1_new).values([{"pk": 1}, {"pk": 2}]))
+
+        rows = collect_orphan_rows("table1", "public.table1_new", "pk", db, schema="public")
+
+        assert rows == [
+            {
+                "ref_table": "table1",
+                "table_name": "public.table2",
+                "schema": "public",
+                "fk_column": "fk",
+                "fk_value": 3,
+                "nb_affected_lines": 1,
+            }
+        ]
+
+    def test_collect_orphan_rows_excludes_tables(self, data):
+        rows = collect_orphan_rows(
+            "table1", "public.table1", "pk", db, schema="public", exclude_tables=["table2"]
+        )
+        assert rows == []

src/utils_flask_sqla/tests/test_revision.py → src/utils_flask_sqla/tests/psql_tests/test_revision.py

@@ -8,86 +8,6 @@
 
 from utils_flask_sqla.revision import alter_table_with_dependent_views
 
-# ---------------------------------------------------------------------------
-# Integration tests against a real PostgreSQL database.
-#
-# These exercise the actual `pg_capture_dependent_views` / `pg_drop_dependent_views`
-# / `pg_recreate_dependent_views` SQL functions (created by migration revision
-# 1d09a9b67970) against a real dependency chain: a table, a materialized view
-# built on that table, and a plain view built on that materialized view.
-#
-# They require a running PostgreSQL server. Point TEST_PG_DATABASE_URI at it,
-# e.g.:
-#   docker run --rm -p 5432:5432 -e POSTGRES_USER=geonatadmin \
-#       -e POSTGRES_PASSWORD=geonatadmin -e POSTGRES_DB=geonature2db \
-#       postgis/postgis:15-3.4
-# Tests are skipped automatically if no server is reachable.
-# ---------------------------------------------------------------------------
-
-PG_URI = os.environ.get(
-    "TEST_PG_DATABASE_URI",
-    "postgresql+psycopg2://geonatadmin:geonatadmin@localhost:5432/geonature2db",
-)
-
-MIGRATION_FILE = (
-    Path(__file__).resolve().parents[1]
-    / "migrations"
-    / "versions"
-    / "1d09a9b67970_add_functions_to_fetch_dependent_view_.py"
-)
-
-
-def _load_migration_module():
-    spec = importlib.util.spec_from_file_location(
-        "utils_flask_sqla_test_dependent_views_migration", MIGRATION_FILE
-    )
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
-@pytest.fixture(scope="module")
-def pg_engine():
-    from alembic.operations import Operations
-    from alembic.runtime.migration import MigrationContext
-
-    engine = create_engine(PG_URI)
-    try:
-        with engine.connect() as conn:
-            conn.execute(text("SELECT 1"))
-    except OperationalError as exc:
-        engine.dispose()
-        pytest.skip(f"PostgreSQL test database not reachable at {PG_URI}: {exc}")
-
-    migration = _load_migration_module()
-    with engine.begin() as conn:
-        with Operations.context(MigrationContext.configure(conn)):
-            migration.upgrade()
-
-    yield engine
-
-    with engine.begin() as conn:
-        with Operations.context(MigrationContext.configure(conn)):
-            migration.downgrade()
-    engine.dispose()
-
-
-@pytest.fixture
-def pg_conn(pg_engine):
-    """A connection wrapping the whole test in one rolled-back transaction.
-
-    PostgreSQL DDL is transactional, so creating the schema/table/views inside
-    this transaction and rolling it back at teardown leaves no trace, without
-    needing per-test unique names.
-    """
-    with pg_engine.connect() as conn:
-        trans = conn.begin()
-        try:
-            yield conn
-        finally:
-            trans.rollback()
-
-
 SCHEMA = "test_revision"