refactor(migrations): simplify calendar schema adoption

src/govoplan_scheduling/backend/migrations/versions/4d5e6f7a8b9c_v018_scheduling_calendar_notifications.py

@@ -6,6 +6,8 @@ Create Date: 2026-07-12 00:00:00.000000
 """
 from __future__ import annotations
 
+from typing import Any
+
 from alembic import op
 import sqlalchemy as sa
 
@@ -55,6 +57,62 @@ _NOTIFICATION_INDEXES = {
 }
 
 
+def _slot_extension_problems(inspector: Any, slot_columns: set[str]) -> list[str]:
+    problems: list[str] = []
+    missing_columns = _CALENDAR_SLOT_COLUMNS - slot_columns
+    if missing_columns:
+        problems.append(
+            "scheduling_candidate_slots missing columns: "
+            f"{', '.join(sorted(missing_columns))}"
+        )
+    indexes = {item["name"] for item in inspector.get_indexes("scheduling_candidate_slots")}
+    missing_indexes = _CALENDAR_SLOT_INDEXES - indexes
+    if missing_indexes:
+        problems.append(
+            "scheduling_candidate_slots missing indexes: "
+            f"{', '.join(sorted(missing_indexes))}"
+        )
+    return problems
+
+
+def _has_cascading_notification_request_foreign_key(inspector: Any) -> bool:
+    return any(
+        tuple(item.get("constrained_columns") or ()) == ("request_id",)
+        and item.get("referred_table") == "scheduling_requests"
+        and tuple(item.get("referred_columns") or ()) == ("id",)
+        and str((item.get("options") or {}).get("ondelete", "")).upper() == "CASCADE"
+        for item in inspector.get_foreign_keys("scheduling_notifications")
+    )
+
+
+def _notification_extension_problems(inspector: Any) -> list[str]:
+    problems: list[str] = []
+    columns = {item["name"] for item in inspector.get_columns("scheduling_notifications")}
+    missing_columns = _NOTIFICATION_COLUMNS - columns
+    if missing_columns:
+        problems.append(
+            "scheduling_notifications missing columns: "
+            f"{', '.join(sorted(missing_columns))}"
+        )
+
+    indexes = {item["name"] for item in inspector.get_indexes("scheduling_notifications")}
+    missing_indexes = _NOTIFICATION_INDEXES - indexes
+    if missing_indexes:
+        problems.append(
+            "scheduling_notifications missing indexes: "
+            f"{', '.join(sorted(missing_indexes))}"
+        )
+
+    primary_key = tuple(
+        inspector.get_pk_constraint("scheduling_notifications").get("constrained_columns") or ()
+    )
+    if primary_key != ("id",):
+        problems.append(f"scheduling_notifications has unexpected primary key: {primary_key!r}")
+    if not _has_cascading_notification_request_foreign_key(inspector):
+        problems.append("scheduling_notifications is missing its cascading request_id foreign key")
+    return problems
+
+
 def _adopt_existing_calendar_notifications() -> bool:
     """Adopt only the complete calendar/notification extension."""
 
@@ -71,59 +129,11 @@ def _adopt_existing_calendar_notifications() -> bool:
     if not present_slot_columns and not notification_exists:
         return False
 
-    problems: list[str] = []
+    problems = _slot_extension_problems(inspector, slot_columns)
-    missing_slot_columns = _CALENDAR_SLOT_COLUMNS - slot_columns
-    if missing_slot_columns:
-        problems.append(
-            "scheduling_candidate_slots missing columns: "
-            f"{', '.join(sorted(missing_slot_columns))}"
-        )
-    slot_indexes = {item["name"] for item in inspector.get_indexes("scheduling_candidate_slots")}
-    missing_slot_indexes = _CALENDAR_SLOT_INDEXES - slot_indexes
-    if missing_slot_indexes:
-        problems.append(
-            "scheduling_candidate_slots missing indexes: "
-            f"{', '.join(sorted(missing_slot_indexes))}"
-        )
-
     if not notification_exists:
         problems.append("missing table: scheduling_notifications")
     else:
-        notification_columns = {
+        problems.extend(_notification_extension_problems(inspector))
-            item["name"] for item in inspector.get_columns("scheduling_notifications")
-        }
-        missing_notification_columns = _NOTIFICATION_COLUMNS - notification_columns
-        if missing_notification_columns:
-            problems.append(
-                "scheduling_notifications missing columns: "
-                f"{', '.join(sorted(missing_notification_columns))}"
-            )
-
-        notification_indexes = {
-            item["name"] for item in inspector.get_indexes("scheduling_notifications")
-        }
-        missing_notification_indexes = _NOTIFICATION_INDEXES - notification_indexes
-        if missing_notification_indexes:
-            problems.append(
-                "scheduling_notifications missing indexes: "
-                f"{', '.join(sorted(missing_notification_indexes))}"
-            )
-
-        primary_key = tuple(
-            inspector.get_pk_constraint("scheduling_notifications").get("constrained_columns") or ()
-        )
-        if primary_key != ("id",):
-            problems.append(f"scheduling_notifications has unexpected primary key: {primary_key!r}")
-
-        request_foreign_key_exists = any(
-            tuple(item.get("constrained_columns") or ()) == ("request_id",)
-            and item.get("referred_table") == "scheduling_requests"
-            and tuple(item.get("referred_columns") or ()) == ("id",)
-            and str((item.get("options") or {}).get("ondelete", "")).upper() == "CASCADE"
-            for item in inspector.get_foreign_keys("scheduling_notifications")
-        )
-        if not request_foreign_key_exists:
-            problems.append("scheduling_notifications is missing its cascading request_id foreign key")
 
     if problems:
         detail = "; ".join(problems)