From fc65e2de4c9c08e7df1886c458b53f7a05894934 Mon Sep 17 00:00:00 2001 From: Fringg Date: Sat, 7 Mar 2026 15:50:31 +0300 Subject: [PATCH] fix: use information_schema for constraint existence checks in migrations Replace pg_class lookup with information_schema.table_constraints query that is schema-qualified and consistent with migration 0028 pattern. Fixes constraint detection on fresh installs where create_all() creates constraints that pg_class lookups could miss. --- .../0015_add_promocode_uses_unique_constraint.py | 12 ++++++++---- ..._add_unique_constraint_transaction_external_id.py | 12 ++++++++---- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/migrations/alembic/versions/0015_add_promocode_uses_unique_constraint.py b/migrations/alembic/versions/0015_add_promocode_uses_unique_constraint.py index 5c79231c..e1a58d39 100644 --- a/migrations/alembic/versions/0015_add_promocode_uses_unique_constraint.py +++ b/migrations/alembic/versions/0015_add_promocode_uses_unique_constraint.py @@ -19,13 +19,17 @@ branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None -def _constraint_exists(constraint_name: str) -> bool: +def _constraint_exists(table: str, constraint_name: str) -> bool: conn = op.get_bind() result = conn.execute( sa.text( - "SELECT 1 FROM pg_class WHERE relname = :name AND relkind IN ('i', 'I')" + "SELECT 1 FROM information_schema.table_constraints " + "WHERE table_schema = current_schema() " + "AND table_name = :table " + "AND constraint_name = :name " + "AND constraint_type = 'UNIQUE'" ), - {'name': constraint_name}, + {'table': table, 'name': constraint_name}, ) return result.scalar() is not None @@ -41,7 +45,7 @@ def upgrade() -> None: ) """) - if not _constraint_exists('uq_promocode_uses_user_promo'): + if not _constraint_exists('promocode_uses', 'uq_promocode_uses_user_promo'): op.create_unique_constraint( 'uq_promocode_uses_user_promo', 'promocode_uses', diff --git a/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py b/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py index 84414de2..0919013d 100644 --- a/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py +++ b/migrations/alembic/versions/0017_add_unique_constraint_transaction_external_id.py @@ -20,13 +20,17 @@ branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None -def _constraint_exists(constraint_name: str) -> bool: +def _constraint_exists(table: str, constraint_name: str) -> bool: conn = op.get_bind() result = conn.execute( sa.text( - "SELECT 1 FROM pg_class WHERE relname = :name AND relkind IN ('i', 'I')" + "SELECT 1 FROM information_schema.table_constraints " + "WHERE table_schema = current_schema() " + "AND table_name = :table " + "AND constraint_name = :name " + "AND constraint_type = 'UNIQUE'" ), - {'name': constraint_name}, + {'table': table, 'name': constraint_name}, ) return result.scalar() is not None @@ -47,7 +51,7 @@ def upgrade() -> None: ) """) - if not _constraint_exists('uq_transaction_external_id_method'): + if not _constraint_exists('transactions', 'uq_transaction_external_id_method'): op.create_unique_constraint( 'uq_transaction_external_id_method', 'transactions',