Source code for

"""Revision command."""
from functools import partial

import alembic

from ...db import signals
from ..alembic import AlembicCommand

[docs]class Revision(AlembicCommand): """Creates an alembic migration revision.""" help = "Create a migration revision"
[docs] def add_arguments(self, parser): parser.add_argument( "args", metavar="app_label", nargs=1, help="Specify the app label to create migrations for." ) parser.add_argument( "-m", "--message", action="store", required=True, help="Message string to use with 'revision'" ) parser.add_argument( "-r", "--rev-id", action="store", help="Specify a hardcoded revision id instead of generating one" ) parser.add_argument( "--head", action="store", help="Specify head revision or <branchname>@head to base new revision on." ) parser.add_argument( "--branch-label", action="store", help="Specify a branch label to apply to the new revision." ) parser.add_argument( "--depends-on", action="store", help="Specify one or more revision identifiers which this revision should depend on.", ) parser.add_argument( "--splice", action="store_true", default=None, help="Allow a non-head revision as the 'head' to splice onto.", ) parser.add_argument( "--autogenerate", action="store_true", dest="autogenerate", help="Populate revision script with candidate migration operations, based on comparison of database to model.", ) parser.add_argument( "--no-autogenerate", action="store_false", dest="autogenerate", default=True, help="Generate blank candidate migration. This will not compare database to models.", )
[docs] def handle( self, app_label, message=None, head=None, splice=None, branch_label=None, depends_on=None, rev_id=None, autogenerate=None, **kwargs, ): appconfig = self.lookup_app(app_label) version_tables = { ".".join( filter( None, [ appconf.config.get_main_option("version_table_schema"), appconf.config.get_main_option("version_table"), ], ) ) for appconf in self.sorcery_apps.values() } @signals.alembic_include_object.connect def include_object(obj=None, name=None, type_=None, reflected=None, compare_to=None): if type_ == "table": return obj in appconfig.tables and obj.fullname not in version_tables else: return obj.table in appconfig.tables command_args = { "autogenerate": autogenerate, "branch_label": branch_label, "depends_on": depends_on, "head": head, "rev_id": rev_id, "message": message, "splice": splice, "sql": False, "version_path": appconfig.version_path, } self.revision_context = alembic.autogenerate.RevisionContext(appconfig.config, appconfig.script, command_args) with alembic.context.EnvironmentContext( appconfig.config, appconfig.script, fn=partial(self.generate_migration, appconfig=appconfig), as_sql=False, template_args=self.revision_context.template_args, revision_context=self.revision_context, ) as context: self.run_env(context, appconfig) for _ in self.revision_context.generate_scripts(): pass # iterate over to generate the migration scripts
[docs] def generate_migration(self, rev, context, appconfig=None): """Generate alembic migration.""" self.revision_context.run_autogenerate(rev, context) return []
Command = Revision