|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +# pylint: disable=no-member |
| 3 | +"""Location: ./mcpgateway/alembic/versions/a54288286395_repair_sqlite_tools_grpc_service_fk.py |
| 4 | +Copyright 2026 |
| 5 | +SPDX-License-Identifier: Apache-2.0 |
| 6 | +Authors: Bogdan Catanus |
| 7 | +
|
| 8 | +repair_sqlite_tools_grpc_service_fk |
| 9 | +
|
| 10 | +Revision ID: a54288286395 |
| 11 | +Revises: 6c0e5f8a9b1d |
| 12 | +Create Date: 2026-06-19 13:05:04.090092 |
| 13 | +
|
| 14 | +Repairs SQLite databases upgraded through revision w7x8y9z0a1b2 that have the |
| 15 | +tools.grpc_service_id column but lack the corresponding foreign key constraint. |
| 16 | +
|
| 17 | +This is a forward-only repair migration that is: |
| 18 | +- Idempotent: safe to run multiple times |
| 19 | +- Database-aware: only operates on SQLite databases |
| 20 | +- Constraint-aware: no-op if FK already exists |
| 21 | +""" |
| 22 | + |
| 23 | +# Standard |
| 24 | +from typing import Sequence, Union |
| 25 | + |
| 26 | +# Third-Party |
| 27 | +from alembic import op |
| 28 | +import sqlalchemy as sa |
| 29 | +from sqlalchemy import inspect |
| 30 | + |
| 31 | +# revision identifiers, used by Alembic. |
| 32 | +revision: str = "a54288286395" # pragma: allowlist secret |
| 33 | +down_revision: Union[str, Sequence[str], None] = "6c0e5f8a9b1d" # pragma: allowlist secret |
| 34 | +branch_labels: Union[str, Sequence[str], None] = None |
| 35 | +depends_on: Union[str, Sequence[str], None] = None |
| 36 | + |
| 37 | +FK_NAME = "fk_tools_grpc_service_id" |
| 38 | + |
| 39 | + |
| 40 | +def _has_foreign_key(inspector: sa.Inspector, table_name: str, fk_name: str) -> bool: |
| 41 | + """Check if a foreign key constraint exists on a table. |
| 42 | +
|
| 43 | + Args: |
| 44 | + inspector: SQLAlchemy inspector instance |
| 45 | + table_name: Name of the table to check |
| 46 | + fk_name: Name of the foreign key constraint |
| 47 | +
|
| 48 | + Returns: |
| 49 | + bool: True if the foreign key exists, False otherwise |
| 50 | + """ |
| 51 | + for fk in inspector.get_foreign_keys(table_name): |
| 52 | + if fk.get("name") == fk_name: |
| 53 | + return True |
| 54 | + return False |
| 55 | + |
| 56 | + |
| 57 | +def upgrade() -> None: |
| 58 | + """Repair missing tools.grpc_service_id FK on SQLite databases. |
| 59 | +
|
| 60 | + This migration: |
| 61 | + 1. Only operates on SQLite (PostgreSQL already has correct FK from w7x8y9z0a1b2) |
| 62 | + 2. Checks if tools table and grpc_service_id column exist |
| 63 | + 3. Checks if FK already exists (idempotent) |
| 64 | + 4. Uses batch_alter_table to rebuild the table with the FK constraint |
| 65 | + """ |
| 66 | + bind = op.get_bind() |
| 67 | + inspector = inspect(bind) |
| 68 | + |
| 69 | + # Only operate on SQLite databases |
| 70 | + if bind.dialect.name != "sqlite": |
| 71 | + return |
| 72 | + |
| 73 | + # Skip if tables don't exist (fresh DB uses db.py models directly) |
| 74 | + if "tools" not in inspector.get_table_names(): |
| 75 | + return |
| 76 | + if "grpc_services" not in inspector.get_table_names(): |
| 77 | + return |
| 78 | + |
| 79 | + # Skip if column doesn't exist (shouldn't happen if w7x8y9z0a1b2 was applied) |
| 80 | + columns = {col["name"] for col in inspector.get_columns("tools")} |
| 81 | + if "grpc_service_id" not in columns: |
| 82 | + return |
| 83 | + |
| 84 | + # Skip if FK already exists (migration already ran or fresh install) |
| 85 | + if _has_foreign_key(inspector, "tools", FK_NAME): |
| 86 | + return |
| 87 | + |
| 88 | + # Repair: rebuild the tools table with the missing FK constraint |
| 89 | + # SQLite requires batch mode with recreate="always" to add FK to existing table |
| 90 | + with op.batch_alter_table("tools", recreate="always") as batch_op: |
| 91 | + batch_op.create_foreign_key( |
| 92 | + FK_NAME, |
| 93 | + "grpc_services", |
| 94 | + ["grpc_service_id"], |
| 95 | + ["id"], |
| 96 | + ondelete="CASCADE", |
| 97 | + ) |
| 98 | + |
| 99 | + |
| 100 | +def downgrade() -> None: |
| 101 | + """Remove the repaired FK constraint on SQLite. |
| 102 | +
|
| 103 | + This is defensive: in practice, downgrading past w7x8y9z0a1b2 should use |
| 104 | + that migration's downgrade logic. This handles the case where a database |
| 105 | + has been repaired by this migration and needs to be rolled back. |
| 106 | + """ |
| 107 | + bind = op.get_bind() |
| 108 | + inspector = inspect(bind) |
| 109 | + |
| 110 | + # Only operate on SQLite databases |
| 111 | + if bind.dialect.name != "sqlite": |
| 112 | + return |
| 113 | + |
| 114 | + # Skip if table doesn't exist |
| 115 | + if "tools" not in inspector.get_table_names(): |
| 116 | + return |
| 117 | + |
| 118 | + # Skip if FK doesn't exist |
| 119 | + if not _has_foreign_key(inspector, "tools", FK_NAME): |
| 120 | + return |
| 121 | + |
| 122 | + # Remove FK constraint (requires table rebuild on SQLite) |
| 123 | + with op.batch_alter_table("tools", recreate="always") as batch_op: |
| 124 | + batch_op.drop_constraint(FK_NAME, type_="foreignkey") |
0 commit comments