# keystone.common.sql.upgrades

# Source code for keystone.common.sql.upgrades

    # Copyright 2013 OpenStack Foundation
    # Copyright 2013 Red Hat, Inc.
    # All Rights Reserved.
    #
    #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    #    not use this file except in compliance with the License. You may obtain
    #    a copy of the License at
    #
    #         http://www.apache.org/licenses/LICENSE-2.0
    #
    #    Unless required by applicable law or agreed to in writing, software
    #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
    #    License for the specific language governing permissions and limitations
    #    under the License.

    import os

    from alembic import command as alembic_api
    from alembic import config as alembic_config
    from alembic import migration as alembic_migration
    from alembic.runtime import environment as alembic_environment
    from alembic import script as alembic_script
    from oslo_db import exception as db_exception
    from oslo_log import log as logging
    from oslo_utils import fileutils

    from keystone.common import sql
    import keystone.conf

    CONF = keystone.conf.CONF
    LOG = logging.getLogger(__name__)

    ALEMBIC_INIT_VERSION = '27e647c0fad4'

    EXPAND_BRANCH = 'expand'
    DATA_MIGRATION_BRANCH = 'data_migration'
    CONTRACT_BRANCH = 'contract'

    RELEASES = ('yoga', 'bobcat', '2024.1', '2025.2', '2026.1')
    MILESTONES = (
        'yoga',
        # Do not add the milestone until the end of the release
    )
    CURRENT_RELEASE = RELEASES[-1]
    MIGRATION_BRANCHES = (EXPAND_BRANCH, CONTRACT_BRANCH)
    VERSIONS_PATH = os.path.join(
        os.path.dirname(sql.__file__), 'migrations', 'versions'
    )

    def get_version_branch_path(release=None, branch=None):
        """Get the path to a version branch."""
        version_path = VERSIONS_PATH
        if branch and release:
            return os.path.join(version_path, release, branch)
        return version_path

    def check_bootstrap_new_branch(branch, version_path, addn_kwargs):
        """Bootstrap a new migration branch if it does not exist."""
        addn_kwargs['version_path'] = version_path
        addn_kwargs['head'] = f'{branch}@head'
        if not os.path.exists(version_path):
            # Bootstrap initial directory structure
            fileutils.ensure_tree(version_path, mode=0o755)

    def _find_alembic_conf():
        """Get the project's alembic configuration.

        :returns: An instance of ``alembic.config.Config``
        """
        path = os.path.join(
            os.path.abspath(os.path.dirname(__file__)), 'alembic.ini'
        )

        config = alembic_config.Config(os.path.abspath(path))

        config.set_main_option('sqlalchemy.url', CONF.database.connection)

        # we don't want to use the logger configuration from the file, which is
        # only really intended for the CLI
        # https://stackoverflow.com/a/42691781/613428
        config.attributes['configure_logger'] = False

        # we want to scan all the versioned subdirectories
        version_paths = [VERSIONS_PATH]
        for release in RELEASES:
            for branch in MIGRATION_BRANCHES:
                version_path = os.path.join(VERSIONS_PATH, release, branch)
                version_paths.append(version_path)
        config.set_main_option('version_locations', os.pathsep.join(version_paths))

        return config

    def get_alembic_config():
        return _find_alembic_conf()

    def _get_current_heads(engine, config):
        script = alembic_script.ScriptDirectory.from_config(config)

        with engine.connect() as conn:
            context = alembic_migration.MigrationContext.configure(conn)
            heads = context.get_current_heads()

        heads_map = _get_head_maps(heads, script)
        return heads_map

    def _get_head_maps(heads, script):
        heads_map = {}

        for head in heads:
            if CONTRACT_BRANCH in script.get_revision(head).branch_labels:
                heads_map[CONTRACT_BRANCH] = head
            else:
                heads_map[EXPAND_BRANCH] = head

        return heads_map

    def get_current_heads():
        """Get the current head of each the expand and contract branches."""
        config = _find_alembic_conf()

        with sql.session_for_read() as session:
            engine = session.get_bind()

        # discard the URL encoded in alembic.ini in favour of the URL
        # configured for the engine by the database fixtures, casting from
        # 'sqlalchemy.engine.url.URL' to str in the process. This returns a
        # RFC-1738 quoted URL, which means that a password like "foo@" will be
        # turned into "foo%40". This in turns causes a problem for
        # set_main_option() because that uses ConfigParser.set, which (by
        # design) uses *python* interpolation to write the string out ... where
        # "%" is the special python interpolation character! Avoid this
        # mismatch by quoting all %'s for the set below.
        engine_url = engine.url.render_as_string(hide_password=False).replace(
            '%', '%%'
        )
        config.set_main_option('sqlalchemy.url', engine_url)

        heads = _get_current_heads(engine, config)

        return heads

    def get_head_revisions():
        """Get the available head for each the expand and contract branches."""
        config = _find_alembic_conf()
        script = alembic_script.ScriptDirectory.from_config(config)
        context = alembic_environment.EnvironmentContext(config, script)
        heads = context.get_head_revisions()
        heads_map = _get_head_maps(heads, script)

        return heads_map

    def _is_database_under_alembic_control(engine):
        with engine.connect() as conn:
            context = alembic_migration.MigrationContext.configure(conn)
            return bool(context.get_current_heads())

    def _upgrade_alembic(engine, config, branch):
        revision = 'heads'
        if branch:
            revision = f'{branch}@head'

        # re-use the connection rather than creating a new one
        with engine.begin() as connection:
            config.attributes['connection'] = connection
            alembic_api.upgrade(config, revision)

    def get_db_version(branch=EXPAND_BRANCH, *, engine=None):
        config = _find_alembic_conf()

        if engine is None:
            with sql.session_for_read() as session:
                engine = session.get_bind()

        # discard the URL encoded in alembic.ini in favour of the URL
        # configured for the engine by the database fixtures, casting from
        # 'sqlalchemy.engine.url.URL' to str in the process. This returns a
        # RFC-1738 quoted URL, which means that a password like "foo@" will be
        # turned into "foo%40". This in turns causes a problem for
        # set_main_option() because that uses ConfigParser.set, which (by
        # design) uses *python* interpolation to write the string out ... where
        # "%" is the special python interpolation character! Avoid this
        # mismatch by quoting all %'s for the set below.
        engine_url = engine.url.render_as_string(hide_password=False).replace(
            '%', '%%'
        )
        config.set_main_option('sqlalchemy.url', engine_url)

        # we use '.get' since the particular branch might not have been created
        alembic_version = _get_current_heads(engine, config).get(branch)

        return alembic_version

    def _db_sync(branch=None, *, engine=None):
        config = _find_alembic_conf()

        if engine is None:
            with sql.session_for_write() as session:
                engine = session.get_bind()

        # discard the URL encoded in alembic.ini in favour of the URL
        # configured for the engine by the database fixtures, casting from
        # 'sqlalchemy.engine.url.URL' to str in the process. This returns a
        # RFC-1738 quoted URL, which means that a password like "foo@" will be
        # turned into "foo%40". This in turns causes a problem for
        # set_main_option() because that uses ConfigParser.set, which (by
        # design) uses *python* interpolation to write the string out ... where
        # "%" is the special python interpolation character! Avoid this
        # mismatch by quoting all %'s for the set below.
        engine_url = engine.url.render_as_string(hide_password=False).replace(
            '%', '%%'
        )
        config.set_main_option('sqlalchemy.url', engine_url)

        _upgrade_alembic(engine, config, branch)

    def _validate_upgrade_order(branch, *, engine=None):
        """Validate the upgrade order of the migration branches.

        This is run before allowing the db_sync command to execute. Ensure the
        expand steps have been run before the contract steps.

        :param branch: The name of the branch that the user is trying to
            upgrade.
        """
        if branch == EXPAND_BRANCH:
            return

        if branch == DATA_MIGRATION_BRANCH:
            # this is a no-op in alembic land
            return

        config = _find_alembic_conf()

        if engine is None:
            with sql.session_for_read() as session:
                engine = session.get_bind()

        script = alembic_script.ScriptDirectory.from_config(config)
        expand_head = None
        for head in script.get_heads():
            if EXPAND_BRANCH in script.get_revision(head).branch_labels:
                expand_head = head
                break

        with engine.connect() as conn:
            context = alembic_migration.MigrationContext.configure(conn)
            current_heads = context.get_current_heads()

        if expand_head not in current_heads:
            raise db_exception.DBMigrationError(
                'You are attempting to upgrade contract ahead of expand. '
                'Please refer to '
                'https://docs.openstack.org/keystone/latest/admin/'
                'identity-upgrading.html '
                'to see the proper steps for rolling upgrades.'
            )

    def expand_schema(engine=None):
        """Expand the database schema ahead of data migration.

        This is run manually by the keystone-manage command before the first
        keystone node is migrated to the latest release.
        """
        _validate_upgrade_order(EXPAND_BRANCH, engine=engine)
        _db_sync(EXPAND_BRANCH, engine=engine)

    def migrate_data(engine=None):
        """Migrate data to match the new schema.

        This is run manually by the keystone-manage command once the keystone
        schema has been expanded for the new release.
        """
        print(
            'Data migrations are no longer supported with alembic. '
            'This is now a no-op.'
        )

    def contract_schema(engine=None):
        """Contract the database.

        This is run manually by the keystone-manage command once the keystone
        nodes have been upgraded to the latest release and will remove any old
        tables/columns that are no longer required.
        """
        _validate_upgrade_order(CONTRACT_BRANCH, engine=engine)
        _db_sync(CONTRACT_BRANCH, engine=engine)

    def offline_sync_database_to_version(version=None, *, engine=None):
        """Perform and off-line sync of the database.

        Migrate the database up to the latest version, doing the equivalent of
        the cycle of --expand, --migrate and --contract, for when an offline
        upgrade is being performed.

        If a version is specified then only migrate the database up to that
        version. Downgrading is not supported. If version is specified, then only
        the main database migration is carried out - and the expand, migration and
        contract phases will NOT be run.
        """
        if version:
            raise Exception('Specifying a version is no longer supported')

        _db_sync(engine=engine)

this page last updated: 2026-08-17 12:45:24

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug