Merge pull request #15811 from jdavcs/dev_migrations_cleanup1

Fix revision scripts, run migrations in CI, add repair option, improve migrations utils
2026-09-24 16:30:27 +08:00 · 2023-04-26 16:09:21 +02:00
parent 47966341c9 380d2b11f4
commit de488f145a
20 changed files with 591 additions and 134 deletions
@@ -56,3 +56,6 @@ jobs:
      - name: Run migration tests
        run: ./run_tests.sh -unit test/unit/data/model/migrations/test_migrations.py
        working-directory: 'galaxy root'
+      - name: Run test migrate database
+        run: ./run_tests.sh -unit test/unit/app/test_migrate_database.py
+        working-directory: 'galaxy root'
@@ -3512,7 +3512,7 @@ class DefaultQuotaAssociation(Base, Dictifiable, RepresentById):
    id = Column(Integer, primary_key=True)
    create_time = Column(DateTime, default=now)
    update_time = Column(DateTime, default=now, onupdate=now)
-    type = Column(String(32), index=True)
+    type = Column(String(32))
    quota_id = Column(Integer, ForeignKey("quota.id"), index=True)
    quota = relationship("Quota", back_populates="default")

@@ -5,13 +5,15 @@ Revises: 6a67bf27e6a6
 Create Date: 2022-06-01 17:50:22.629894

 """
-from alembic import op
 from sqlalchemy import (
    Column,
    DateTime,
 )

-from galaxy.model.migrations.util import drop_column
+from galaxy.model.migrations.util import (
+    add_column,
+    drop_column,
+)
 from galaxy.model.orm.now import now

 # revision identifiers, used by Alembic.
@@ -29,4 +31,4 @@ def upgrade():


 def downgrade():
-    op.add_column(table_name, Column("update_time", DateTime, default=now, onupdate=now))
+    add_column(table_name, Column(column_name, DateTime, default=now, onupdate=now))
@@ -5,7 +5,6 @@ Revises: 518c8438a91b
 Create Date: 2023-01-13 16:13:09.578391

 """
-from alembic import op
 from sqlalchemy import (
    Column,
    ForeignKey,
@@ -14,6 +13,10 @@ from sqlalchemy import (
 )

 from galaxy.model.custom_types import TrimmedString
+from galaxy.model.migrations.util import (
+    create_table,
+    drop_table,
+)

 # revision identifiers, used by Alembic.
 revision = "3100452fa030"
@@ -27,7 +30,7 @@ table_name = "workflow_invocation_message"


 def upgrade():
-    op.create_table(
+    create_table(
        table_name,
        Column("id", Integer, primary_key=True),
        Column("reason", String(32)),
@@ -43,4 +46,4 @@ def upgrade():


 def downgrade():
-    op.drop_table(table_name)
+    drop_table(table_name)
@@ -26,4 +26,4 @@ def upgrade():


 def downgrade():
-    drop_index(index_name, table_name, columns)
+    drop_index(index_name, table_name)
@@ -5,12 +5,11 @@ Revises: 59e024ceaca1
 Create Date: 2022-10-24 16:43:39.565871

 """
-import sqlalchemy as sa
-from alembic import op
+from sqlalchemy import Column

 from galaxy.model.custom_types import JSONType
 from galaxy.model.migrations.util import (
-    column_exists,
+    add_column,
    drop_column,
 )

@@ -26,8 +25,7 @@ column_name = "when_expression"


 def upgrade():
-    if not column_exists(table_name, column_name):
-        op.add_column(table_name, sa.Column(column_name, JSONType))
+    add_column(table_name, Column(column_name, JSONType))


 def downgrade():
@@ -5,7 +5,6 @@ Revises: e0e3bb173ee6
 Create Date: 2022-10-12 18:02:34.659770

 """
-from alembic import op
 from sqlalchemy import (
    Column,
    DateTime,
@@ -17,6 +16,10 @@ from galaxy.model.custom_types import (
    TrimmedString,
    UUIDType,
 )
+from galaxy.model.migrations.util import (
+    create_table,
+    drop_table,
+)

 # revision identifiers, used by Alembic.
 revision = "59e024ceaca1"
@@ -29,7 +32,7 @@ table_name = "store_export_association"


 def upgrade():
-    op.create_table(
+    create_table(
        table_name,
        Column("id", Integer, primary_key=True),
        Column("task_uuid", UUIDType(), index=True, unique=True),
@@ -41,4 +44,4 @@ def upgrade():


 def downgrade():
-    op.drop_table(table_name)
+    drop_table(table_name)
@@ -5,13 +5,16 @@ Revises: b182f655505f
 Create Date: 2022-03-14 12:17:55.313830

 """
-from alembic import op
 from sqlalchemy import (
    Boolean,
    Column,
 )

-from galaxy.model.migrations.util import drop_column
+from galaxy.model.migrations.util import (
+    add_column,
+    drop_column,
+    transaction,
+)

 # revision identifiers, used by Alembic.
 revision = "6a67bf27e6a6"
@@ -21,10 +24,12 @@ depends_on = None


 def upgrade():
-    op.add_column("history_dataset_association", Column("metadata_deferred", Boolean(), default=False))
-    op.add_column("library_dataset_dataset_association", Column("metadata_deferred", Boolean(), default=False))
+    with transaction():
+        add_column("history_dataset_association", Column("metadata_deferred", Boolean(), default=False))
+        add_column("library_dataset_dataset_association", Column("metadata_deferred", Boolean(), default=False))


 def downgrade():
-    drop_column("history_dataset_association", "metadata_deferred")
-    drop_column("library_dataset_dataset_association", "metadata_deferred")
+    with transaction():
+        drop_column("history_dataset_association", "metadata_deferred")
+        drop_column("library_dataset_dataset_association", "metadata_deferred")
@@ -5,14 +5,17 @@ Revises: d0583094c8cd
 Create Date: 2022-06-10 10:38:25.212102

 """
-from alembic import op
 from sqlalchemy import (
    Column,
    String,
 )

 from galaxy.model.custom_types import JSONType
-from galaxy.model.migrations.util import drop_column
+from galaxy.model.migrations.util import (
+    add_column,
+    drop_column,
+    transaction,
+)

 # revision identifiers, used by Alembic.
 revision = "9540a051226e"
@@ -22,15 +25,17 @@ depends_on = None


 def upgrade():
-    preferred_object_store_type = String(255)
-    op.add_column("galaxy_user", Column("preferred_object_store_id", preferred_object_store_type, default=None))
-    op.add_column("history", Column("preferred_object_store_id", preferred_object_store_type, default=None))
-    op.add_column("job", Column("preferred_object_store_id", preferred_object_store_type, default=None))
-    op.add_column("job", Column("object_store_id_overrides", JSONType))
+    with transaction():
+        preferred_object_store_type = String(255)
+        add_column("galaxy_user", Column("preferred_object_store_id", preferred_object_store_type, default=None))
+        add_column("history", Column("preferred_object_store_id", preferred_object_store_type, default=None))
+        add_column("job", Column("preferred_object_store_id", preferred_object_store_type, default=None))
+        add_column("job", Column("object_store_id_overrides", JSONType))


 def downgrade():
-    drop_column("galaxy_user", "preferred_object_store_id")
-    drop_column("history", "preferred_object_store_id")
-    drop_column("job", "preferred_object_store_id")
-    drop_column("job", "object_store_id_overrides")
+    with transaction():
+        drop_column("galaxy_user", "preferred_object_store_id")
+        drop_column("history", "preferred_object_store_id")
+        drop_column("job", "preferred_object_store_id")
+        drop_column("job", "object_store_id_overrides")
@@ -5,12 +5,11 @@ Revises: e7b6dcb09efd
 Create Date: 2022-03-14 12:56:57.067748

 """
-from alembic import op
 from sqlalchemy import Column

 from galaxy.model.custom_types import JSONType
 from galaxy.model.migrations.util import (
-    column_exists,
+    add_column,
    drop_column,
 )

@@ -26,8 +25,7 @@ column_name = "source_metadata"


 def upgrade():
-    if not column_exists(table_name, column_name):
-        op.add_column(table_name, Column(column_name, JSONType))
+    add_column(table_name, Column(column_name, JSONType))


 def downgrade():
@@ -7,6 +7,8 @@ Create Date: 2023-01-16 11:53:59.783836
 """
 from alembic import op

+from galaxy.model.migrations.util import transaction
+
 # revision identifiers, used by Alembic.
 revision = "c39f1de47a04"
 down_revision = "3100452fa030"
@@ -88,14 +90,12 @@ GROUP BY jobstates.hdca_id


 def upgrade():
-    op.execute("BEGIN")
-    op.execute(f"DROP VIEW IF EXISTS {view_name}")
-    op.execute(f"CREATE VIEW {view_name} AS {new_aggregate_query}")
-    op.execute("END")
+    with transaction():
+        op.execute(f"DROP VIEW IF EXISTS {view_name}")
+        op.execute(f"CREATE VIEW {view_name} AS {new_aggregate_query}")


 def downgrade():
-    op.execute("BEGIN")
-    op.execute(f"DROP VIEW IF EXISTS {view_name}")
-    op.execute(f"CREATE VIEW {view_name} AS {previous_aggregate_query}")
-    op.execute("END")
+    with transaction():
+        op.execute(f"DROP VIEW IF EXISTS {view_name}")
+        op.execute(f"CREATE VIEW {view_name} AS {previous_aggregate_query}")
@@ -27,4 +27,4 @@ def upgrade():


 def downgrade():
-    drop_index(index_name, table_name, columns)
+    drop_index(index_name, table_name)
@@ -5,7 +5,6 @@ Revises: c39f1de47a04
 Create Date: 2022-06-09 12:24:44.329038

 """
-from alembic import op
 from sqlalchemy import (
    Column,
    ForeignKey,
@@ -15,8 +14,15 @@ from sqlalchemy import (
 )

 from galaxy.model.migrations.util import (
-    add_unique_constraint,
+    add_column,
+    create_index,
+    create_table,
+    create_unique_constraint,
    drop_column,
+    drop_constraint,
+    drop_index,
+    drop_table,
+    transaction,
 )

 # revision identifiers, used by Alembic.
@@ -27,23 +33,27 @@ depends_on = None


 def upgrade():
-    op.add_column("quota", Column("quota_source_label", String(32), default=None))
-
-    op.create_table(
-        "user_quota_source_usage",
-        Column("id", Integer, primary_key=True),
-        Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True),
-        Column("quota_source_label", String(32), index=True),
-        # user had an index on disk_usage - does that make any sense? -John
-        Column("disk_usage", Numeric(15, 0)),
-    )
-    add_unique_constraint("uqsu_unique_label_per_user", "user_quota_source_usage", ["user_id", "quota_source_label"])
-    op.drop_index("ix_default_quota_association_type", "default_quota_association")
-    op.create_index("ix_quota_quota_source_label", "quota", ["quota_source_label"])
+    with transaction():
+        add_column("quota", Column("quota_source_label", String(32), default=None))
+        create_table(
+            "user_quota_source_usage",
+            Column("id", Integer, primary_key=True),
+            Column("user_id", Integer, ForeignKey("galaxy_user.id"), index=True),
+            Column("quota_source_label", String(32), index=True),
+            # user had an index on disk_usage - does that make any sense? -John
+            Column("disk_usage", Numeric(15, 0)),
+        )
+        create_unique_constraint(
+            "uqsu_unique_label_per_user", "user_quota_source_usage", ["user_id", "quota_source_label"]
+        )
+        drop_index("ix_default_quota_association_type", "default_quota_association")
+        create_index("ix_quota_quota_source_label", "quota", ["quota_source_label"])


 def downgrade():
-    op.create_index("ix_default_quota_association_type", "default_quota_association", ["type"], unique=True)
-    op.drop_table("user_quota_source_usage")
-    op.drop_index("ix_quota_quota_source_label", "quota")
-    drop_column("quota", "quota_source_label")
+    with transaction():
+        drop_index("ix_quota_quota_source_label", "quota")
+        create_index("ix_default_quota_association_type", "default_quota_association", ["type"], unique=True)
+        drop_constraint("uqsu_unique_label_per_user", "user_quota_source_usage")
+        drop_table("user_quota_source_usage")
+        drop_column("quota", "quota_source_label")
@@ -5,15 +5,16 @@ Revises: 186d4835587b
 Create Date: 2022-09-27 14:09:05.890227

 """
-from alembic import op
 from sqlalchemy import (
    Boolean,
    Column,
 )

 from galaxy.model.migrations.util import (
-    column_exists,
+    add_column,
    drop_column,
+    drop_index,
+    transaction,
 )

 # revision identifiers, used by Alembic.
@@ -29,9 +30,10 @@ column_name = "deleted"


 def upgrade():
-    if not column_exists(table_name, column_name):
-        op.add_column(table_name, Column(column_name, Boolean(), default=False))
+    add_column(table_name, Column(column_name, Boolean(), default=False, index=True))


 def downgrade():
-    drop_column(table_name, column_name)
+    with transaction():
+        drop_index("ix_api_keys_deleted", table_name)
+        drop_column(table_name, column_name)
@@ -58,21 +58,23 @@ class BaseParserBuilder(abc.ABC):
        self.subparsers = parser.add_subparsers(required=subcommand_required)
        self._init_arg_parsers()

-    def add_upgrade_command(self):
+    def add_upgrade_command(self, dev_options=False):
+        parents = self._get_upgrade_downgrade_arg_parsers(dev_options)
        parser = self._add_parser(
            "upgrade",
            self._cmd.upgrade,
            "Upgrade to a later version",
-            parents=[self._sql_arg_parser],
+            parents=parents,
        )
        parser.add_argument("revision", help="Revision identifier or release tag", nargs="?")

-    def add_downgrade_command(self):
+    def add_downgrade_command(self, dev_options=False):
+        parents = self._get_upgrade_downgrade_arg_parsers(dev_options)
        parser = self._add_parser(
            "downgrade",
            self._cmd.downgrade,
            "Revert to a previous version",
-            parents=[self._sql_arg_parser],
+            parents=parents,
        )
        parser.add_argument("revision", help="Revision identifier or release tag")

@@ -131,6 +133,13 @@ class BaseParserBuilder(abc.ABC):
    def _init_arg_parsers(self):
        self._verbose_arg_parser = self._make_verbose_arg_parser()
        self._sql_arg_parser = self._make_sql_arg_parser()
+        self._repair_arg_parser = self._make_repair_arg_parser()
+
+    def _get_upgrade_downgrade_arg_parsers(self, dev_options):
+        parsers = [self._sql_arg_parser]
+        if dev_options:
+            parsers.append(self._make_repair_arg_parser())
+        return parsers

    def _make_verbose_arg_parser(self):
        parser = ArgumentParser(add_help=False)
@@ -146,6 +155,20 @@ class BaseParserBuilder(abc.ABC):
        )
        return parser

+    def _make_repair_arg_parser(self):
+        parser = ArgumentParser(add_help=False)
+        parser.add_argument(
+            "--repair",
+            action="store_true",
+            help="""Skip revisions that conflict with the current state of the database (examples of
+            conflict: creating an object that exists or dropping an object that does not exist).
+            Note: implicitly created objects (such as those created by Alembic as part of ALTER
+            statement workaround) that may have been left over will still raise an error. Such
+            objects must be removed manually.
+            """,
+        )
+        return parser
+
    def _add_parser(self, command, func, help, aliases=None, parents=None):
        aliases = aliases or []
        parents = parents or []
@@ -174,6 +197,7 @@ class BaseDbScript(abc.ABC):
        command.revision(self.alembic_config, message=args.message, rev_id=args.rev_id, head=head)

    def upgrade(self, args: Namespace) -> None:
+        self._process_repair_arg(args)
        if args.revision:
            revision = self._parse_revision(args.revision)
            self._upgrade_to_revision(revision, args.sql)
@@ -181,6 +205,7 @@ class BaseDbScript(abc.ABC):
            self._upgrade_to_head(args.sql)

    def downgrade(self, args: Namespace) -> None:
+        self._process_repair_arg(args)
        revision = self._parse_revision(args.revision)
        command.downgrade(self.alembic_config, revision, args.sql)

@@ -227,6 +252,10 @@ class BaseDbScript(abc.ABC):
        # Subclasses that have revision tags should overwrite this method.
        return {}

+    def _process_repair_arg(self, args: Namespace) -> None:
+        if "repair" in args and args.repair:
+            self.alembic_config.set_main_option("repair", "1")
+

 class BaseCommand(abc.ABC):
    @abc.abstractmethod
@@ -1,81 +1,368 @@
 import logging
-from typing import List
+from abc import (
+    ABC,
+    abstractmethod,
+)
+from contextlib import contextmanager
+from typing import (
+    Any,
+    List,
+    Optional,
+    Sequence,
+)

+import sqlalchemy as sa
 from alembic import (
    context,
    op,
 )
-from sqlalchemy import inspect
+from sqlalchemy.exc import OperationalError

 log = logging.getLogger(__name__)


-def drop_column(table_name, column_name):
+class DDLOperation(ABC):
+    """Base class for all DDL operations."""
+
+    def run(self) -> Optional[Any]:
+        if not self._is_repair_mode():
+            return self.execute()
+        else:
+            if self.pre_execute_check():
+                return self.execute()
+            else:
+                self.log_check_not_passed()
+                return None
+
+    @abstractmethod
+    def execute(self) -> Optional[Any]:
+        ...
+
+    @abstractmethod
+    def pre_execute_check(self) -> bool:
+        ...
+
+    @abstractmethod
+    def log_check_not_passed(self) -> None:
+        ...
+
+    def _is_repair_mode(self) -> bool:
+        """`--repair` option has been passed to the command."""
+        return bool(context.config.get_main_option("repair"))
+
+    def _log_object_exists_message(self, object_name: str) -> None:
+        log.info(f"{object_name} already exists. Skipping revision.")
+
+    def _log_object_does_not_exist_message(self, object_name: str) -> None:
+        log.info(f"{object_name} does not exist. Skipping revision.")
+
+
+class DDLAlterOperation(DDLOperation):
+    """
+    Base class for DDL operations that implement special handling of ALTER statements.
+    Ref:
+    - https://alembic.sqlalchemy.org/en/latest/ops.html#alembic.operations.Operations.batch_alter_table
+    - https://alembic.sqlalchemy.org/en/latest/batch.html
+    """
+
+    def __init__(self, table_name: str) -> None:
+        self.table_name = table_name
+
+    def run(self) -> Optional[Any]:
+        if context.is_offline_mode():
+            log.info("Generation of `alter` statements is disabled in offline mode.")
+            return None
+        return super().run()
+
+    def execute(self) -> Optional[Any]:
+        if _is_sqlite():
+            with legacy_alter_table(), op.batch_alter_table(self.table_name) as batch_op:
+                return self.batch_execute(batch_op)
+        else:
+            return self.non_batch_execute()  # use regular op context for non-sqlite db
+
+    @abstractmethod
+    def batch_execute(self, batch_op) -> Optional[Any]:
+        ...
+
+    @abstractmethod
+    def non_batch_execute(self) -> Optional[Any]:
+        ...
+
+
+class CreateTable(DDLOperation):
+    """Wraps alembic's create_table directive."""
+
+    def __init__(self, table_name: str, *columns: sa.schema.SchemaItem) -> None:
+        self.table_name = table_name
+        self.columns = columns
+
+    def execute(self) -> Optional[sa.Table]:
+        return op.create_table(self.table_name, *self.columns)
+
+    def pre_execute_check(self) -> bool:
+        return not table_exists(self.table_name, False)
+
+    def log_check_not_passed(self) -> None:
+        self._log_object_exists_message(f"{self.table_name} table")
+
+
+class DropTable(DDLOperation):
+    """Wraps alembic's drop_table directive."""
+
+    def __init__(self, table_name: str) -> None:
+        self.table_name = table_name
+
+    def execute(self) -> None:
+        op.drop_table(self.table_name)
+
+    def pre_execute_check(self) -> bool:
+        return table_exists(self.table_name, False)
+
+    def log_check_not_passed(self) -> None:
+        self._log_object_does_not_exist_message(f"{self.table_name} table")
+
+
+class CreateIndex(DDLOperation):
+    """Wraps alembic's create_index directive."""
+
+    def __init__(self, index_name: str, table_name: str, columns: Sequence, **kw: Any) -> None:
+        self.index_name = index_name
+        self.table_name = table_name
+        self.columns = columns
+        self.kw = kw
+
+    def execute(self) -> None:
+        op.create_index(self.index_name, self.table_name, self.columns, **self.kw)
+
+    def pre_execute_check(self) -> bool:
+        return not index_exists(self.index_name, self.table_name, False)
+
+    def log_check_not_passed(self) -> None:
+        name = _table_object_description(self.index_name, self.table_name)
+        self._log_object_exists_message(name)
+
+
+class DropIndex(DDLOperation):
+    """Wraps alembic's drop_index directive."""
+
+    def __init__(self, index_name: str, table_name: str) -> None:
+        self.index_name = index_name
+        self.table_name = table_name
+
+    def execute(self) -> None:
+        op.drop_index(self.index_name, self.table_name)
+
+    def pre_execute_check(self) -> bool:
+        return index_exists(self.index_name, self.table_name, False)
+
+    def log_check_not_passed(self) -> None:
+        name = _table_object_description(self.index_name, self.table_name)
+        self._log_object_does_not_exist_message(name)
+
+
+class AddColumn(DDLOperation):
+    """Wraps alembic's add_column directive."""
+
+    def __init__(self, table_name: str, column: sa.Column) -> None:
+        self.table_name = table_name
+        self.column = column
+
+    def execute(self) -> None:
+        op.add_column(self.table_name, self.column)
+
+    def pre_execute_check(self) -> bool:
+        return not column_exists(self.table_name, self.column.name, False)
+
+    def log_check_not_passed(self) -> None:
+        name = _table_object_description(self.column.name, self.table_name)
+        self._log_object_exists_message(name)
+
+
+class DropColumn(DDLAlterOperation):
+    """Wraps alembic's drop_column directive."""
+
+    def __init__(self, table_name: str, column_name: str) -> None:
+        super().__init__(table_name)
+        self.column_name = column_name
+
+    def batch_execute(self, batch_op) -> None:
+        batch_op.drop_column(self.column_name)
+
+    def non_batch_execute(self) -> None:
+        op.drop_column(self.table_name, self.column_name)
+
+    def pre_execute_check(self) -> bool:
+        return column_exists(self.table_name, self.column_name, False)
+
+    def log_check_not_passed(self) -> None:
+        name = _table_object_description(self.column_name, self.table_name)
+        self._log_object_does_not_exist_message(name)
+
+
+class CreateUniqueConstraint(DDLAlterOperation):
+    """Wraps alembic's create_unique_constraint directive."""
+
+    def __init__(self, constraint_name: str, table_name: str, columns: List[str]) -> None:
+        super().__init__(table_name)
+        self.constraint_name = constraint_name
+        self.columns = columns
+
+    def batch_execute(self, batch_op) -> None:
+        batch_op.create_unique_constraint(self.constraint_name, self.columns)
+
+    def non_batch_execute(self) -> None:
+        op.create_unique_constraint(self.constraint_name, self.table_name, self.columns)
+
+    def pre_execute_check(self) -> bool:
+        return not unique_constraint_exists(self.constraint_name, self.table_name, False)
+
+    def log_check_not_passed(self) -> None:
+        name = _table_object_description(self.constraint_name, self.table_name)
+        self._log_object_exists_message(name)
+
+
+class DropConstraint(DDLAlterOperation):
+    """Wraps alembic's drop_constraint directive."""
+
+    def __init__(self, constraint_name: str, table_name: str) -> None:
+        super().__init__(table_name)
+        self.constraint_name = constraint_name
+
+    def batch_execute(self, batch_op) -> None:
+        batch_op.drop_constraint(self.constraint_name)
+
+    def non_batch_execute(self) -> None:
+        op.drop_constraint(self.constraint_name, self.table_name)
+
+    def pre_execute_check(self) -> bool:
+        return unique_constraint_exists(self.constraint_name, self.table_name, False)
+
+    def log_check_not_passed(self) -> None:
+        name = _table_object_description(self.constraint_name, self.table_name)
+        self._log_object_does_not_exist_message(name)
+
+
+def create_table(table_name: str, *columns: sa.schema.SchemaItem) -> Optional[sa.Table]:
+    return CreateTable(table_name, *columns).run()
+
+
+def drop_table(table_name: str) -> None:
+    DropTable(table_name).run()
+
+
+def add_column(table_name: str, column: sa.Column) -> None:
+    AddColumn(table_name, column).run()
+
+
+def drop_column(table_name, column_name) -> None:
+    DropColumn(table_name, column_name).run()
+
+
+def create_index(index_name, table_name, columns, **kw) -> None:
+    CreateIndex(index_name, table_name, columns, **kw).run()
+
+
+def drop_index(index_name, table_name) -> None:
+    DropIndex(index_name, table_name).run()
+
+
+def create_unique_constraint(constraint_name: str, table_name: str, columns: List[str]) -> None:
+    CreateUniqueConstraint(constraint_name, table_name, columns).run()
+
+
+def drop_constraint(constraint_name: str, table_name: str) -> None:
+    DropConstraint(constraint_name, table_name).run()
+
+
+def table_exists(table_name: str, default: bool) -> bool:
+    """Check if table exists. If running in offline mode, return default."""
    if context.is_offline_mode():
-        return _handle_offline_mode(f"drop_column({table_name}, {column_name})", None)
-
-    with op.batch_alter_table(table_name) as batch_op:
-        batch_op.drop_column(column_name)
+        _log_offline_mode_message(table_exists.__name__, default)
+        return default
+    return _inspector().has_table(table_name)


-def add_unique_constraint(index_name: str, table_name: str, columns: List[str]):
-    if _is_sqlite():
-        with op.batch_alter_table(table_name) as batch_op:
-            batch_op.create_unique_constraint(index_name, columns)
-    else:
-        op.create_unique_constraint(index_name, table_name, columns)
-
-
-def drop_unique_constraint(index_name: str, table_name: str):
-    if _is_sqlite():
-        with op.batch_alter_table(table_name) as batch_op:
-            batch_op.drop_constraint(index_name)
-    else:
-        op.drop_constraint(index_name, table_name)
-
-
-def create_index(index_name, table_name, columns):
-    if index_exists(index_name, table_name):
-        msg = f"Index with name {index_name} on {table_name} already exists. Skipping revision."
-        log.info(msg)
-    else:
-        op.create_index(index_name, table_name, columns)
-
-
-def drop_index(index_name, table_name, columns):
-    if index_exists(index_name, table_name):
-        op.drop_index(index_name, table_name)
-
-
-def column_exists(table_name, column_name):
+def column_exists(table_name: str, column_name: str, default: bool) -> bool:
+    """Check if column exists. If running in offline mode, return default."""
    if context.is_offline_mode():
-        return _handle_offline_mode(f"column_exists({table_name}, {column_name})")
-
-    bind = op.get_context().bind
-    insp = inspect(bind)
-    columns = insp.get_columns(table_name)
+        _log_offline_mode_message(column_exists.__name__, default)
+        return default
+    columns = _inspector().get_columns(table_name)
    return any(c["name"] == column_name for c in columns)


-def index_exists(index_name, table_name):
+def index_exists(index_name: str, table_name: str, default: bool) -> bool:
+    """Check if index exists. If running in offline mode, return default."""
    if context.is_offline_mode():
-        return _handle_offline_mode(f"index_exists({index_name}, {table_name})")
-
-    bind = op.get_context().bind
-    insp = inspect(bind)
-    indexes = insp.get_indexes(table_name)
+        _log_offline_mode_message(index_exists.__name__, default)
+        return default
+    indexes = _inspector().get_indexes(table_name)
    return any(index["name"] == index_name for index in indexes)


-def _handle_offline_mode(code, return_value=False):
-    msg = (
-        "This script is being executed in offline mode and cannot connect to the database. "
-        f"Therefore, `{code}` returns `{return_value}` by default."
+def unique_constraint_exists(constraint_name: str, table_name: str, default: bool) -> bool:
+    """Check if unique constraint exists. If running in offline mode, return default."""
+    if context.is_offline_mode():
+        _log_offline_mode_message(unique_constraint_exists.__name__, default)
+        return default
+    constraints = _inspector().get_unique_constraints(table_name)
+    return any(c["name"] == constraint_name for c in constraints)
+
+
+def _table_object_description(object_name: str, table_name: str) -> str:
+    return f"{object_name} on {table_name} table"
+
+
+def _log_offline_mode_message(function_name: str, return_value: Any) -> None:
+    log.info(
+        f"This script is being executed in offline mode, so it cannot connect to the database. "
+        f"Therefore, function `{function_name}` will return the value `{return_value}`, "
+        f"which is the expected value during normal operation."
    )
-    log.info(msg)
-    return return_value
+
+
+def _inspector() -> Any:
+    bind = op.get_context().bind
+    return sa.inspect(bind)


 def _is_sqlite() -> bool:
    bind = op.get_context().bind
    return bool(bind and bind.engine.name == "sqlite")
+
+
+@contextmanager
+def legacy_alter_table():
+    """
+    Wrapper required for add/drop column statements.
+    Prevents error when column belongs to a table referenced in a view. Relevant to sqlite only.
+    Ref: https://github.com/sqlalchemy/alembic/issues/1207
+    Ref: https://sqlite.org/pragma.html#pragma_legacy_alter_table
+    """
+    try:
+        op.execute("PRAGMA legacy_alter_table=1;")
+        yield
+    finally:
+        op.execute("PRAGMA legacy_alter_table=0;")
+
+
+@contextmanager
+def transaction():
+    """
+    Wraps multiple statements in upgrade/downgrade revision script functions in
+    a database transaction, ensuring transactional control.
+
+    Used for SQLite only. Although SQLite supports transactional DDL, pysqlite does not.
+    Ref: https://bugs.python.org/issue10740
+    """
+    if not _is_sqlite():
+        yield  # For postgresql, alembic ensures transactional context.
+    else:
+        try:
+            op.execute("BEGIN")
+            yield
+            op.execute("END")
+        except OperationalError:
+            op.execute("ROLLBACK")
+            raise
@@ -45,7 +45,7 @@ def create_and_drop_database(url: DbUrl) -> Iterator[None]:
        create_database(url)
        yield
    finally:
-        if _is_postgres(url):
+        if is_postgres(url):
            _drop_postgres_database(url)


@@ -58,7 +58,7 @@ def drop_existing_database(url: DbUrl) -> Iterator[None]:
    try:
        yield
    finally:
-        if _is_postgres(url):
+        if is_postgres(url):
            _drop_postgres_database(url)


@@ -72,6 +72,20 @@ def disposing_engine(url: DbUrl) -> Iterator[Engine]:
        engine.dispose()


+@pytest.fixture
+def sqlite_url_factory(tmp_directory: str) -> Callable[[], DbUrl]:
+    """
+    Same as url_factory, except this returns a sqlite url only.
+    This is used when we want to ensure a test runs under sqlite.
+    """
+
+    def url() -> DbUrl:
+        database = _generate_unique_database_name()
+        return _make_sqlite_db_url(tmp_directory, database)
+
+    return url
+
+
@pytest.fixture
 def url_factory(tmp_directory: str) -> Callable[[], DbUrl]:
    """
@@ -127,7 +141,7 @@ def drop_database(db_url, database):

    Used only for test purposes to cleanup after creating a test database.
    """
-    if _is_postgres(db_url) or _is_mysql(db_url):
+    if is_postgres(db_url) or _is_mysql(db_url):
        _drop_database(db_url, database)
    else:
        url = make_url(db_url)
@@ -207,7 +221,7 @@ def get_stored_instance_by_id(session, cls_, id):
    return session.execute(statement).scalar_one()


-def _is_postgres(url: DbUrl) -> bool:
+def is_postgres(url: DbUrl) -> bool:
    return url.startswith("postgres")


@@ -23,8 +23,8 @@ def main() -> None:

    parser_builder = ParserBuilder(parser)

-    parser_builder.add_upgrade_command()
-    parser_builder.add_downgrade_command()
+    parser_builder.add_upgrade_command(dev_options=True)
+    parser_builder.add_downgrade_command(dev_options=True)
    parser_builder.add_version_command()
    parser_builder.add_dbversion_command()
    parser_builder.add_init_command()
@@ -304,6 +304,14 @@ class TestUpgradeCommand:
        heads = get_db_heads(config)
        assert heads == ("d",)

+    def test_repair_arg_available_to_dev_script_only(self, config, command):
+        completed = run_command(f"{command} upgrade --repair")
+        if command == DEV_CMD:
+            assert completed.returncode == 0
+        else:
+            assert completed.returncode == 2
+            assert "unrecognized arguments: --repair" in completed.stderr
+

@pytest.mark.parametrize("command", COMMANDS)
 class TestDowngradeCommand:
@@ -364,3 +372,11 @@ class TestDowngradeCommand:

        heads = get_db_heads(config)
        assert "a" in heads
+
+    def test_repair_arg_available_to_dev_script_only(self, config, command):
+        completed = run_command(f"{command} downgrade base --repair")
+        if command == DEV_CMD:
+            assert completed.returncode == 0
+        else:
+            assert completed.returncode == 2
+            assert "unrecognized arguments: --repair" in completed.stderr
@@ -0,0 +1,82 @@
+import pytest
+
+from galaxy.model.migrations import AlembicManager
+from galaxy.model.unittest_utils.migration_scripts_testing_utils import (  # noqa: F401 - contains fixtures we have to import explicitly
+    run_command,
+    tmp_directory,
+)
+from galaxy.model.unittest_utils.model_testing_utils import (  # noqa: F401 - url_factory is a fixture we have to import explicitly
+    create_and_drop_database,
+    disposing_engine,
+    is_postgres,
+    sqlite_url_factory,
+    url_factory,
+)
+
+
+@pytest.fixture(params=["sqlite", "postgres"])
+def dburl(request, url_factory, sqlite_url_factory):  # noqa: F811
+    if request.param == "sqlite":
+        return sqlite_url_factory()
+    else:
+        dburl = url_factory()
+        if is_postgres(dburl):
+            return dburl
+
+
+def test_migrations(dburl, monkeypatch):
+    """
+    Verify that database migration revision scripts can be applied in the following sequence:
+        1. `init` (initialize a new database from model definition)
+        2. `downgrade base` (downgrade to pre-alembic state)
+        3. `upgrade` (upgrade to up-to-date after downgrading)
+        4. `downgrade base` (downgrade to pre-alembic state after upgradingx)
+    Note that step "2" is not the same as step "4": the state of the database preceeding "2" is a result of
+    initializing it from the model definition; the state of the database in "3" is a a result of running
+    all the revision scripts in sequence from base to the current head (up-to-date state).
+
+    If a postgresql connection is available, this text will be executed twice: against sqlite and postgresql.
+    Otherwise, the test will run once against a sqlite database.
+    The database is created and destroyed within the scope of this test function.
+    """
+    if not dburl:
+        # If a postgresql url is not available, dburl will be None on the second run of the test.
+        return
+
+    command = "manage_db.sh"
+    monkeypatch.setenv("GALAXY_CONFIG_OVERRIDE_DATABASE_CONNECTION", dburl)
+    monkeypatch.setenv("GALAXY_INSTALL_CONFIG_OVERRIDE_INSTALL_DATABASE_CONNECTION", dburl)
+
+    with create_and_drop_database(dburl), disposing_engine(dburl) as engine:
+        # Step 1. Initialize database from model definition.
+        completed = run_command(f"{command} init")
+        assert completed.returncode == 0
+        assert "Running stamp_revision" in completed.stderr
+        dbversion_after_init = _get_database_version(engine)
+        assert dbversion_after_init
+
+        # Running `upgrade` next would be a noop, since the `init` command "stamps"
+        # the alembic_version table with the latest revision.
+        # Step 2. Downgrade to base (verify downgrading runs without errors)
+        completed = run_command(f"{command} downgrade base")
+        assert completed.returncode == 0
+        assert "Running downgrade" in completed.stderr
+        assert not _get_database_version(engine)
+
+        # Step 3. Upgrade to current head (verify upgrading runs without errors)
+        completed = run_command(f"{command} upgrade")
+        assert completed.returncode == 0
+        assert "Running upgrade" in completed.stderr
+        dbversion_after_upgrade = _get_database_version(engine)
+        assert dbversion_after_upgrade == dbversion_after_init
+
+        # Step 4. Downgrade to base (verify downgrading after upgrading runs without errors)
+        completed = run_command(f"{command} downgrade base")
+        assert completed.returncode == 0
+        assert "Running downgrade" in completed.stderr
+        assert not _get_database_version(engine)
+
+
+def _get_database_version(engine):
+    alembic_manager = AlembicManager(engine)
+    return alembic_manager.db_heads