<s> from __future__ import print_function <EOL> from copy import copy , deepcopy <EOL> import datetime <EOL> import inspect <EOL> import sys <EOL> import traceback <EOL> from django . core . management import call_command <EOL> from django . core . management . commands import loaddata <EOL> from django . db import models <EOL> import south . db <EOL> from south import exceptions <EOL> from south . db import DEFAULT_DB_ALIAS <EOL> from south . models import MigrationHistory <EOL> from south . signals import ran_migration <EOL> from south . utils . py3 import StringIO <EOL> class Migrator ( object ) : <EOL> def __init__ ( self , verbosity = <NUM_LIT:0> , interactive = False ) : <EOL> self . verbosity = int ( verbosity ) <EOL> self . interactive = bool ( interactive ) <EOL> @ staticmethod <EOL> def title ( target ) : <EOL> raise NotImplementedError ( ) <EOL> def print_title ( self , target ) : <EOL> if self . verbosity : <EOL> print ( self . title ( target ) ) <EOL> @ staticmethod <EOL> def status ( target ) : <EOL> raise NotImplementedError ( ) <EOL> def print_status ( self , migration ) : <EOL> status = self . status ( migration ) <EOL> if self . verbosity and status : <EOL> print ( status ) <EOL> @ staticmethod <EOL> def orm ( migration ) : <EOL> raise NotImplementedError ( ) <EOL> def backwards ( self , migration ) : <EOL> return self . _wrap_direction ( migration . backwards ( ) , migration . prev_orm ( ) ) <EOL> def direction ( self , migration ) : <EOL> raise NotImplementedError ( ) <EOL> @ staticmethod <EOL> def _wrap_direction ( direction , orm ) : <EOL> args = inspect . getargspec ( direction ) <EOL> if len ( args [ <NUM_LIT:0> ] ) == <NUM_LIT:1> : <EOL> return direction <EOL> return ( lambda : direction ( orm ) ) <EOL> @ staticmethod <EOL> def record ( migration , database ) : <EOL> raise NotImplementedError ( ) <EOL> def run_migration_error ( self , migration , extra_info = '<STR_LIT>' ) : <EOL> return ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT:\n>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) % extra_info <EOL> def run_migration ( self , migration , database ) : <EOL> migration_function = self . direction ( migration ) <EOL> south . db . db . start_transaction ( ) <EOL> try : <EOL> migration_function ( ) <EOL> south . db . db . execute_deferred_sql ( ) <EOL> if not isinstance ( getattr ( self , '<STR_LIT>' , self ) , DryRunMigrator ) : <EOL> self . record ( migration , database ) <EOL> except : <EOL> south . db . db . rollback_transaction ( ) <EOL> if not south . db . db . has_ddl_transactions : <EOL> print ( self . run_migration_error ( migration ) ) <EOL> print ( "<STR_LIT>" % migration ) <EOL> raise <EOL> else : <EOL> try : <EOL> south . db . db . commit_transaction ( ) <EOL> except : <EOL> print ( "<STR_LIT>" % migration ) <EOL> raise <EOL> def run ( self , migration , database ) : <EOL> south . db . db . current_orm = self . orm ( migration ) <EOL> if not isinstance ( getattr ( self , '<STR_LIT>' , self ) , DryRunMigrator ) : <EOL> if not south . db . db . has_ddl_transactions : <EOL> dry_run = DryRunMigrator ( migrator = self , ignore_fail = False ) <EOL> dry_run . run_migration ( migration , database ) <EOL> return self . run_migration ( migration , database ) <EOL> def send_ran_migration ( self , migration , database ) : <EOL> ran_migration . send ( None , <EOL> app = migration . app_label ( ) , <EOL> migration = migration , <EOL> method = self . __class__ . __name__ . lower ( ) , <EOL> verbosity = self . verbosity , <EOL> interactive = self . interactive , <EOL> db = database ) <EOL> def migrate ( self , migration , database ) : <EOL> """<STR_LIT>""" <EOL> app = migration . migrations . _migrations <EOL> migration_name = migration . name ( ) <EOL> self . print_status ( migration ) <EOL> result = self . run ( migration , database ) <EOL> self . send_ran_migration ( migration , database ) <EOL> return result <EOL> def migrate_many ( self , target , migrations , database ) : <EOL> raise NotImplementedError ( ) <EOL> class MigratorWrapper ( object ) : <EOL> def __init__ ( self , migrator , * args , ** kwargs ) : <EOL> self . _migrator = copy ( migrator ) <EOL> attributes = dict ( [ ( k , getattr ( self , k ) ) <EOL> for k in self . __class__ . __dict__ <EOL> if not k . startswith ( '<STR_LIT>' ) ] ) <EOL> self . _migrator . __dict__ . update ( attributes ) <EOL> self . _migrator . __dict__ [ '<STR_LIT>' ] = self <EOL> def __getattr__ ( self , name ) : <EOL> return getattr ( self . _migrator , name ) <EOL> class DryRunMigrator ( MigratorWrapper ) : <EOL> def __init__ ( self , ignore_fail = True , * args , ** kwargs ) : <EOL> super ( DryRunMigrator , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _ignore_fail = ignore_fail <EOL> def _run_migration ( self , migration ) : <EOL> if migration . no_dry_run ( ) : <EOL> if self . verbosity : <EOL> print ( "<STR_LIT>" % migration ) <EOL> return <EOL> south . db . db . dry_run = True <EOL> constraint_cache = deepcopy ( south . db . db . _constraint_cache ) <EOL> if self . _ignore_fail : <EOL> south . db . db . debug , old_debug = False , south . db . db . debug <EOL> pending_creates = south . db . db . get_pending_creates ( ) <EOL> south . db . db . start_transaction ( ) <EOL> migration_function = self . direction ( migration ) <EOL> try : <EOL> try : <EOL> migration_function ( ) <EOL> south . db . db . execute_deferred_sql ( ) <EOL> except : <EOL> raise exceptions . FailedDryRun ( migration , sys . exc_info ( ) ) <EOL> finally : <EOL> south . db . db . rollback_transactions_dry_run ( ) <EOL> if self . _ignore_fail : <EOL> south . db . db . debug = old_debug <EOL> south . db . db . clear_run_data ( pending_creates ) <EOL> south . db . db . dry_run = False <EOL> south . db . db . _constraint_cache = constraint_cache <EOL> def run_migration ( self , migration , database ) : <EOL> try : <EOL> self . _run_migration ( migration ) <EOL> except exceptions . FailedDryRun : <EOL> if self . _ignore_fail : <EOL> return False <EOL> raise <EOL> def send_ran_migration ( self , * args , ** kwargs ) : <EOL> pass <EOL> class FakeMigrator ( MigratorWrapper ) : <EOL> def run ( self , migration , database ) : <EOL> self . record ( migration , database ) <EOL> if self . verbosity : <EOL> print ( '<STR_LIT>' ) <EOL> def send_ran_migration ( self , * args , ** kwargs ) : <EOL> pass <EOL> class LoadInitialDataMigrator ( MigratorWrapper ) : <EOL> def load_initial_data ( self , target , db = '<STR_LIT:default>' ) : <EOL> if target is None or target != target . migrations [ - <NUM_LIT:1> ] : <EOL> return <EOL> if self . verbosity : <EOL> print ( "<STR_LIT>" % target . app_label ( ) ) <EOL> old_get_apps = models . get_apps <EOL> new_get_apps = lambda : [ models . get_app ( target . app_label ( ) ) ] <EOL> models . get_apps = new_get_apps <EOL> loaddata . get_apps = new_get_apps <EOL> try : <EOL> call_command ( '<STR_LIT>' , '<STR_LIT>' , verbosity = self . verbosity , database = db ) <EOL> finally : <EOL> models . get_apps = old_get_apps <EOL> loaddata . get_apps = old_get_apps <EOL> def migrate_many ( self , target , migrations , database ) : <EOL> migrator = self . _migrator <EOL> result = migrator . __class__ . migrate_many ( migrator , target , migrations , database ) <EOL> if result : <EOL> self . load_initial_data ( target , db = database ) <EOL> return True <EOL> class Forwards ( Migrator ) : <EOL> """<STR_LIT>""" <EOL> torun = '<STR_LIT>' <EOL> @ staticmethod <EOL> def title ( target ) : <EOL> if target is not None : <EOL> return "<STR_LIT>" % target . name ( ) <EOL> else : <EOL> assert False , "<STR_LIT>" <EOL> @ staticmethod <EOL> def status ( migration ) : <EOL> return '<STR_LIT>' % migration <EOL> @ staticmethod <EOL> def orm ( migration ) : <EOL> return migration . orm ( ) <EOL> def forwards ( self , migration ) : <EOL> return self . _wrap_direction ( migration . forwards ( ) , migration . orm ( ) ) <EOL> direction = forwards <EOL> @ staticmethod <EOL> def record ( migration , database ) : <EOL> record = MigrationHistory . for_migration ( migration , database ) <EOL> try : <EOL> from django . utils . timezone import now <EOL> record . applied = now ( ) <EOL> except ImportError : <EOL> record . applied = datetime . datetime . utcnow ( ) <EOL> if database != DEFAULT_DB_ALIAS : <EOL> record . save ( using = database ) <EOL> else : <EOL> record . save ( ) <EOL> def format_backwards ( self , migration ) : <EOL> if migration . no_dry_run ( ) : <EOL> return "<STR_LIT>" <EOL> old_debug , old_dry_run = south . db . db . debug , south . db . db . dry_run <EOL> south . db . db . debug = south . db . db . dry_run = True <EOL> stdout = sys . stdout <EOL> sys . stdout = StringIO ( ) <EOL> try : <EOL> try : <EOL> self . backwards ( migration ) ( ) <EOL> return sys . stdout . getvalue ( ) <EOL> except : <EOL> raise <EOL> finally : <EOL> south . db . db . debug , south . db . db . dry_run = old_debug , old_dry_run <EOL> sys . stdout = stdout <EOL> def run_migration_error ( self , migration , extra_info = '<STR_LIT>' ) : <EOL> extra_info = ( '<STR_LIT:\n>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT:%s>' <EOL> '<STR_LIT:%s>' % <EOL> ( self . format_backwards ( migration ) , extra_info ) ) <EOL> return super ( Forwards , self ) . run_migration_error ( migration , extra_info ) <EOL> def migrate_many ( self , target , migrations , database ) : <EOL> try : <EOL> for migration in migrations : <EOL> result = self . migrate ( migration , database ) <EOL> if result is False : <EOL> return False <EOL> finally : <EOL> south . db . db . send_pending_create_signals ( verbosity = self . verbosity , <EOL> interactive = self . interactive ) <EOL> return True <EOL> class Backwards ( Migrator ) : <EOL> """<STR_LIT>""" <EOL> torun = '<STR_LIT>' <EOL> @ staticmethod <EOL> def title ( target ) : <EOL> if target is None : <EOL> return "<STR_LIT>" <EOL> else : <EOL> return "<STR_LIT>" % target . name ( ) <EOL> @ staticmethod <EOL> def status ( migration ) : <EOL> return '<STR_LIT>' % migration <EOL> @ staticmethod <EOL> def orm ( migration ) : <EOL> return migration . prev_orm ( ) <EOL> direction = Migrator . backwards <EOL> @ staticmethod <EOL> def record ( migration , database ) : <EOL> record = MigrationHistory . for_migration ( migration , database ) <EOL> if record . id is not None : <EOL> if database != DEFAULT_DB_ALIAS : <EOL> record . delete ( using = database ) <EOL> else : <EOL> record . delete ( ) <EOL> def migrate_many ( self , target , migrations , database ) : <EOL> for migration in migrations : <EOL> self . migrate ( migration , database ) <EOL> return True </s>
