db: Adds alembic scaffolding for DB migrations

oslo_db has deprecated sqlalchemy-migrate in favor of alembic, and the
old facade no longer exists in current oslo.db releases.

This adds the alembic config, env, and template scaffolding as the first
step of porting coriolis-dbsync to alembic.

The implementation is based on OpenStack Nova and Neutron.

coriolis/db/sqlalchemy/alembic/README.rst

@@ -0,0 +1,16 @@
+Migrations for the main database
+================================
+
+This directory contains migrations for the coriolis database. These are implemented
+using `alembic`__, a lightweight database migration tool designed for usage
+with `SQLAlchemy`__.
+
+The best place to start understanding Alembic is with its own `tutorial`__. You
+can also play around with the :command:`alembic` command::
+
+    $ alembic --help
+
+.. __: https://alembic.sqlalchemy.org/en/latest/
+.. __: https://www.sqlalchemy.org/
+.. __: https://alembic.sqlalchemy.org/en/latest/tutorial.html
+

coriolis/db/sqlalchemy/alembic/alembic.ini

@@ -0,0 +1,49 @@
+# A generic, single database configuration.
+
+[alembic]
+# path to migration scripts
+script_location = %(here)s
+
+# template used to generate migration files
+# file_template = %%(rev)s_%%(slug)s
+
+# default to an empty string because the migration cli will
+# extract the correct value and set it programmatically before alembic is fully
+# invoked.
+sqlalchemy.url =
+path_separator = space
+
+# Logging configuration
+[loggers]
+keys = root,sqlalchemy,alembic
+
+[handlers]
+keys = console
+
+[formatters]
+keys = generic
+
+[logger_root]
+level = WARN
+handlers = console
+qualname =
+
+[logger_sqlalchemy]
+level = WARN
+handlers =
+qualname = sqlalchemy.engine
+
+[logger_alembic]
+level = INFO
+handlers =
+qualname = alembic
+
+[handler_console]
+class = StreamHandler
+args = (sys.stderr,)
+level = NOTSET
+formatter = generic
+
+[formatter_generic]
+format = %(levelname)-5.5s [%(name)s] %(message)s
+datefmt = %H:%M:%S

coriolis/db/sqlalchemy/alembic/env.py

@@ -0,0 +1,80 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+from alembic import context
+from sqlalchemy import engine_from_config
+from sqlalchemy import pool
+
+from coriolis.db.sqlalchemy import models
+
+# this is the Alembic Config object, which provides
+# access to the values within the .ini file in use.
+config = context.config
+
+# this is the MetaData object for the various models in the database.
+target_metadata = models.BASE.metadata
+
+
+def run_migrations_offline():
+    """Run migrations in 'offline' mode.
+
+    This configures the context with just a URL and not an Engine, though an
+    Engine is acceptable here as well.  By skipping the Engine creation we
+    don't even need a DBAPI to be available.
+
+    Calls to context.execute() here emit the given string to the script output.
+    """
+    url = config.get_main_option("sqlalchemy.url")
+    context.configure(
+        url=url,
+        target_metadata=target_metadata,
+        render_as_batch=True,
+        literal_binds=True,
+        dialect_opts={"paramstyle": "named"},
+    )
+
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+def run_migrations_online():
+    """Run migrations in 'online' mode.
+
+    In this scenario we need to create an Engine and associate a connection
+    with the context.
+    """
+    connectable = config.attributes.get("connection", None)
+
+    if connectable is not None:
+        context.configure(
+            connection=connectable,
+            target_metadata=target_metadata,
+            render_as_batch=True,
+        )
+
+        with context.begin_transaction():
+            context.run_migrations()
+        return
+
+    # only create Engine if we don't have a Connection from the outside.
+    connectable = engine_from_config(
+        config.get_section(config.config_ini_section),
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
+
+    with connectable.connect() as connection:
+        context.configure(
+            connection=connection,
+            target_metadata=target_metadata,
+            render_as_batch=True,
+        )
+
+        with context.begin_transaction():
+            context.run_migrations()
+
+
+if context.is_offline_mode():
+    run_migrations_offline()
+else:
+    run_migrations_online()

coriolis/db/sqlalchemy/alembic/script.py.mako

@@ -0,0 +1,27 @@
+# Copyright ${create_date.year} Cloudbase Solutions Srl
+# All Rights Reserved.
+
+"""${message}
+
+Revision ID: ${up_revision}
+Revises: ${down_revision | comma,n}
+Create Date: ${create_date}
+"""
+
+from alembic import op
+import sqlalchemy
+${imports if imports else ""}
+
+# revision identifiers, used by Alembic.
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = ${repr(branch_labels)}
+depends_on = ${repr(depends_on)}
+
+
+def upgrade():
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade():
+    ${downgrades if downgrades else "pass"}

coriolis/tests/db/sqlalchemy/alembic/test_env.py

@@ -0,0 +1,82 @@
+# Copyright 2026 Cloudbase Solutions Srl
+# All Rights Reserved.
+
+import importlib
+import sys
+from unittest import mock
+
+from coriolis.tests import test_base
+
+ENV_MODULE_NAME = "coriolis.db.sqlalchemy.alembic.env"
+
+
+class AlembicEnvTestCase(test_base.CoriolisBaseTestCase):
+    """Test suite for the Coriolis Alembic 'env.py' migration script."""
+
+    def setUp(self):
+        super(AlembicEnvTestCase, self).setUp()
+        self.addCleanup(sys.modules.pop, ENV_MODULE_NAME, None)
+
+    def _import_env(self, offline_mode, connection=None):
+        # run_migrations_online or run_migrations_offline runs on module
+        # import, so we need to set up what we need beforehand.
+        sys.modules.pop(ENV_MODULE_NAME, None)
+
+        mock_context = mock.MagicMock()
+        mock_context.is_offline_mode.return_value = offline_mode
+        mock_context.config.attributes.get.return_value = connection
+        with mock.patch("alembic.context", mock_context):
+            env = importlib.import_module(ENV_MODULE_NAME)
+
+        return env, mock_context
+
+    def test_offline_mode(self):
+        env, mock_context = self._import_env(offline_mode=True)
+
+        self.assertIs(env.config, mock_context.config)
+        mock_context.config.get_main_option.assert_called_once_with(
+            "sqlalchemy.url")
+        mock_context.configure.assert_called_once_with(
+            url=mock_context.config.get_main_option.return_value,
+            target_metadata=env.target_metadata,
+            render_as_batch=True,
+            literal_binds=True,
+            dialect_opts={"paramstyle": "named"},
+        )
+        mock_context.begin_transaction.assert_called_once_with()
+        mock_context.run_migrations.assert_called_once_with()
+
+    def test_online_mode_with_existing_connection(self):
+        env, mock_context = self._import_env(
+            offline_mode=False, connection=mock.sentinel.connection)
+
+        mock_context.config.attributes.get.assert_called_once_with(
+            "connection", None)
+        mock_context.configure.assert_called_once_with(
+            connection=mock.sentinel.connection,
+            target_metadata=env.target_metadata,
+            render_as_batch=True,
+        )
+        mock_context.begin_transaction.assert_called_once_with()
+        mock_context.run_migrations.assert_called_once_with()
+
+    @mock.patch("sqlalchemy.engine_from_config")
+    def test_online_mode_creates_engine(self, mock_engine_from_config):
+        env, mock_context = self._import_env(offline_mode=False)
+
+        mock_engine_from_config.assert_called_once_with(
+            mock_context.config.get_section.return_value,
+            prefix="sqlalchemy.",
+            poolclass=env.pool.NullPool,
+        )
+        mock_connectable = mock_engine_from_config.return_value
+        mock_connectable.connect.assert_called_once_with()
+        mock_connection = (
+            mock_connectable.connect.return_value.__enter__.return_value)
+        mock_context.configure.assert_called_once_with(
+            connection=mock_connection,
+            target_metadata=env.target_metadata,
+            render_as_batch=True,
+        )
+        mock_context.begin_transaction.assert_called_once_with()
+        mock_context.run_migrations.assert_called_once_with()

requirements.txt

@@ -1,4 +1,5 @@
 setuptools>=65.0.0,<82  # pkg_resources removed in 82; required by sqlalchemy-migrate
+alembic
 keystoneauth1
 keystonemiddleware
 Jinja2