From 74e4508b8fab5aaa0c14196c3cc5d3761e5b8c5f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 12:59:29 +0200 Subject: [PATCH] refactor(migrations): simplify calendar schema adoption --- ..._v018_scheduling_calendar_notifications.py | 110 ++++++++++-------- 1 file changed, 60 insertions(+), 50 deletions(-) diff --git a/src/govoplan_scheduling/backend/migrations/versions/4d5e6f7a8b9c_v018_scheduling_calendar_notifications.py b/src/govoplan_scheduling/backend/migrations/versions/4d5e6f7a8b9c_v018_scheduling_calendar_notifications.py index 9b4df07..f0022e7 100644 --- a/src/govoplan_scheduling/backend/migrations/versions/4d5e6f7a8b9c_v018_scheduling_calendar_notifications.py +++ b/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] = [] - 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))}" - ) - + problems = _slot_extension_problems(inspector, slot_columns) if not notification_exists: problems.append("missing table: scheduling_notifications") else: - notification_columns = { - 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") + problems.extend(_notification_extension_problems(inspector)) if problems: detail = "; ".join(problems)