Main migration logic.
from django.core.exceptions import ImproperlyConfigured
from south import exceptions
from south.models import MigrationHistory
from south.db import db, DEFAULT_DB_ALIAS
from south.migration.migrators import (Backwards, Forwards,
from south.migration.base import Migration, Migrations
from south.migration.utils import SortedSet
from south.migration.base import all_migrations
from south.signals import pre_migrate, post_migrate
def to_apply(forwards, done):
return [m for m in forwards if m not in done]
def to_unapply(backwards, done):
return [m for m in backwards if (m in done) or m.is_rebase()]
def problems(pending, done):
last = None
if not pending:
for migration in pending:
if migration in done:
last = migration
if last and migration not in done:
yield last, migration
def forwards_problems(pending, done, verbosity):
Takes the list of linearised pending migrations, and the set of done ones,
and returns the list of problems, if any.
return inner_problem_check(problems(reversed(pending), done), done, verbosity)
def backwards_problems(pending, done, verbosity):
return inner_problem_check(problems(pending, done), done, verbosity)
def inner_problem_check(problems, done, verbosity):
"Takes a set of possible problems and gets the actual issues out of it."
result = 
for last, migration in problems:
# 'Last' is the last applied migration. Step back from it until we
# either find nothing wrong, or we find something.
to_check = list(last.dependencies)
checking = to_check.pop()
if checking not in done:
# That's bad. Error.
print (" ! Migration %s should not have been applied "
"before %s but was." % (last, checking))
def check_migration_histories(histories, delete_ghosts=False, ignore_ghosts=False):
"Checks that there's no 'ghost' migrations in the database."
exists = SortedSet()
ghosts = 
for h in histories:
m = h.get_migration()
pass # Ignore missing applications
# They may want us to delete ghosts.
for h in ghosts:
elif not ignore_ghosts:
def get_dependencies(target, migrations):
forwards = lambda allow_rebase=False: list()
backwards = lambda allow_rebase=False: list()
if target is None:
backwards = migrations.backwards_plan
forwards = target.forwards_plan
# When migrating backwards we want to remove up to and
# including the next migration up in this app (not the next
# one, that includes other apps)
migration_before_here = target.next()
backwards = migration_before_here.backwards_plan
return forwards, backwards
def get_direction(target, applied, migrations, verbosity, interactive):
# Get the forwards and reverse dependencies for this target
forwards, backwards = get_dependencies(target, migrations)
# Is the whole forward branch applied?
problems = None
if target is not None:
forwards = forwards(allow_rebase = not [x for x in applied if x.migrations is target.migrations])
workplan = to_apply(forwards, applied)
workplan = None
# See if there's some forward work to be done.
if not workplan:
# If they're all applied, we only know it's not backwards
direction = None
# If the remaining migrations are strictly a right segment of
# the forwards trace, we just need to go forwards to our
# target (and check for badness)
problems = forwards_problems(forwards, applied, verbosity)
direction = Forwards(verbosity=verbosity, interactive=interactive)
if not problems:
# What about the whole backward trace then?
backwards = backwards(allow_rebase=target is None)
missing_backwards = to_apply(backwards, applied)
if missing_backwards != backwards:
# If what's missing is a strict left segment of backwards (i.e.
# all the higher migrations) then we need to go backwards
workplan = to_unapply(backwards, applied)
problems = backwards_problems(
[x for x in backwards if not x.is_rebase()],
direction = Backwards(verbosity=verbosity, interactive=interactive)
return direction, problems, workplan
def get_migrator(direction, db_dry_run, fake, load_initial_data):
if not direction:
direction = DryRunMigrator(migrator=direction, ignore_fail=False)
direction = FakeMigrator(migrator=direction)
direction = LoadInitialDataMigrator(migrator=direction)
def migrate_app(migrations, target_name=None, merge=False, fake=False, db_dry_run=False, yes=False, verbosity=0, load_initial_data=False, skip=False, database=DEFAULT_DB_ALIAS, delete_ghosts=False, ignore_ghosts=False, interactive=False):
app_label = migrations.app_label()
verbosity = int(verbosity)
# Fire off the pre-migrate signal
# If there aren't any, quit quizically
if not migrations:
print "? You have no migrations for the '%s' app. You might want some." % app_label
# Load the entire dependency graph
# Check there's no strange ones in the database
applied = MigrationHistory.objects.filter(applied__isnull=False)
# If we're using a different database, use that
if database != DEFAULT_DB_ALIAS:
applied = applied.using(database)
south.db.db = south.db.dbs[database]
# We now have to make sure the migrations are all reloaded, as they'll
# have imported the old value of south.db.db.
south.db.db.debug = (verbosity > 1)
applied = check_migration_histories(applied, delete_ghosts, ignore_ghosts)
# Guess the target_name
target = migrations.guess_migration(target_name)
if target_name not in ('zero', None) and target.name() != target_name:
print " - Soft matched migration %s to %s." % (target_name,
print "Running migrations for %s:" % app_label
# Get the forwards and reverse dependencies for this target
direction, problems, workplan = get_direction(target, applied, migrations,
if problems and not (merge or skip):
# Perform the migration
migrator = get_migrator(direction, db_dry_run, fake, load_initial_data)
success = migrator.migrate_many(target, workplan, database)
# Finally, fire off the post-migrate signal
# Say there's nothing.
print '- Nothing to migrate.'
# If we have initial data enabled, and we're at the most recent
# migration, do initial data.
# Note: We use a fake Forwards() migrator here. It's never used really.
migrator = LoadInitialDataMigrator(migrator=Forwards(verbosity=verbosity))
# Send signal.