Source code for

"""Upgrade command."""
from functools import partial

import alembic
from import CommandError

from ..alembic import AlembicCommand

[docs]class Upgrade(AlembicCommand): """Apply upgrade migration revisions.""" help = "Apply migration revisions"
[docs] def add_arguments(self, parser): parser.add_argument("app_label", nargs="?", help="App label of application to limit the output to.") parser.add_argument( "-r", "--revision", default="heads", help="Database state will be brought to the state after that " 'migration. Use the name "heads" to apply all migrations.', ) parser.add_argument( "-s", "--sql", default=False, action="store_true", help="Don't emit SQL to database - dump to standard output/file instead", )
[docs] def handle(self, app_label=None, revision=None, sql=False, **kwargs): appconfigs = [self.lookup_app(app_label)] if app_label is not None else self.sorcery_apps.values() starting_rev = None if ":" in revision: starting_rev, revision = revision.split(":", 2) if revision != "heads" and app_label is None: raise CommandError("Revision requires an app_label to be provided") for appconfig in appconfigs: if not sql: self.stdout.write( "Running migrations for {} on database {}".format(, appconfig.db.alias) ) ) with alembic.context.EnvironmentContext( appconfig.config, appconfig.script, fn=partial(self.upgrade, appconfig=appconfig, revision=revision), as_sql=sql, starting_rev=starting_rev, destination_rev=revision, ) as context: self.run_env(context, appconfig)
[docs] def upgrade(self, rev, context, appconfig, revision): """Executes alembic upgrade revisions to the given revision.""" return appconfig.script._upgrade_revs(revision, rev)
Command = Upgrade