A custom manage.py file for managing pending migrations (only)

manage_custom.py

__author__ = "Parag Tyagi"

# set environment
import os
import sys
import django
sys.path.append('../')
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings')
django.setup()

from django.core.management import execute_from_command_line
from django.db import DEFAULT_DB_ALIAS, connections
from django.db.migrations.executor import MigrationExecutor


class Migration(object):
    """
    A custom manage.py file for managing pending migrations (only)
    """

    def __init__(self, migrate_per_migration_id=False):
        """
        :param migrate_per_migration_id: Setting this to `True` will migrate each pending migration of any 
        particular app individually. `False` will migrate the whole app at a time.
        
        You can add more arguments (viz. showmigrations, migrate) by defining the argument with prefix as 'ARGV_'
        and create its functionality accordingly.
        """
        self.ARG_PREFIX = 'ARGV_'
        self.MIGRATE_PER_MIGRATION_ID = migrate_per_migration_id
        self.ARGV_showmigrations = False
        self.ARGV_migrate = False

    @staticmethod
    def get_pending_migrations(database):
        """
        :param database: Database alias
        :return: List of pending migrations
        """
        connection = connections[database]
        connection.prepare_database()
        executor = MigrationExecutor(connection)
        targets = executor.loader.graph.leaf_nodes()
        return executor.migration_plan(targets)

    def check_arguments(self, args):
        """
        Method for checking arguments passed while running the command
        :param args: Dictionary of arguments passed while running the script file
        :return: Set the argument variable ('ARGV_<argument>') to True if found else terminate the script
        """
        required_args = filter(None, [var.split(self.ARG_PREFIX)[1] if var.startswith(self.ARG_PREFIX)
                                      else None for var in self.__dict__.keys()])
        if any(k in args for k in required_args):
            for arg in required_args:
                if arg in args:
                    setattr(self, '{}{}'.format(self.ARG_PREFIX, arg), True)
                    break
        else:
            print ("Please pass argument: {}"
                   "\ne.g. python manage_custom.py {}".format(required_args, required_args[0]))
            sys.exit()

    def do_migration(self):
        """
        Migrates all the pending migrations (if any)
        """
        pending_migrations = self.get_pending_migrations(DEFAULT_DB_ALIAS)
        if pending_migrations:
            done_app = []
            for mig in pending_migrations:
                app, migration_id = str(mig[0]).split('.')
                commands = ['manage.py', 'migrate'] + ([app, migration_id] if self.MIGRATE_PER_MIGRATION_ID else [app])
                if self.ARGV_migrate and (app not in done_app or self.MIGRATE_PER_MIGRATION_ID):
                    execute_from_command_line(commands)
                    done_app.append(app)
                elif self.ARGV_showmigrations:
                    print (str(mig[0]))
        else:
            print ("No pending migrations")


if __name__ == '__main__':
    args = sys.argv
    migration = Migration()
    migration.check_arguments(args)
    migration.do_migration()

Usage:

# below command will show all pending migrations
python manage_custom.py showmigrations

# below command will migrate all pending migrations
python manage_custom.py migrate