Проверка ожидающих миграции Django

В Django существует ли простой способ проверить, были ли запущены все миграции баз данных? Я нашел manage.py migrate --list, который дает мне информацию, которую я хочу, но формат не очень читается машиной.

В контексте: у меня есть script, который не должен запускаться до тех пор, пока база данных не будет перенесена. По разным причинам было бы сложно отправить сигнал из процесса, выполняющего миграцию. Поэтому я бы хотел, чтобы мой script периодически проверял базу данных, чтобы проверить, все ли запущены миграции.

Ответы

Ответ 1

Shell

Единственное простое решение, которое я нашел, - это запустить

./manage.py showmigrations | grep '\[ \]'

который выведет пустую строку в случае применения всех миграций.

Однако он тесно связан с форматом вывода.

Python

Я проверил исходный код команды migrate, и похоже, что это должно сработать:

from django.db.migrations.executor import MigrationExecutor
from django.db import connections, DEFAULT_DB_ALIAS


def is_database_synchronized(database):
    connection = connections[database]
    connection.prepare_database()
    executor = MigrationExecutor(connection)
    targets = executor.loader.graph.leaf_nodes()
    return not executor.migration_plan(targets)

# Usage example.
if is_database_synchronized(DEFAULT_DB_ALIAS):
    # All migrations have been applied.
    pass
else:
    # Unapplied migrations found.
    pass

Ответ 2

1.10 примечания к выпуску:

Новый параметр makemigrations --check делает вывод команды с ненулевым статусом при обнаружении изменений модели без миграции.

Ответ 3

Пытаться,

python manage.py migrate --list | grep "\[ \]\|^[a-z]" | grep "[ ]" -B 1

возвращается,

<app_1>
 [ ] 0001_initial
 [ ] 0002_auto_01201244
 [ ] 0003_auto_12334333

<app_2>
 [ ] 0031_auto_12344544
 [ ] 0032_auto_45456767
 [ ] 0033_auto_23346566

<app_3>
 [ ] 0008_auto_3446677

Обновление:

Если вы обновили версию Django> = 1.11, используйте команду ниже,

python manage.py showmigrations | grep '\[ \]\|^[a-z]' | grep '[  ]' -B 1

Ответ 4

./manage.py showmigrations #check, какие уже выполненные миграции были применены или нет
(или: ./manage.py showmigrations someApp # только для конкретного приложения)

./manage.py makemigrations --dry-run #check для выполнения миграций
(или: ./manage.py makemigrations someApp --dry-run # только для конкретного приложения)

./manage.py makemigrations сделать миграцию
(или: ./manage.py makemigrations someApp # только для конкретного приложения)

./manage.py sqlmigrate someApp 0001 # просмотр изменений SQL для конкретного приложения и миграции

./manage.py migrate применить миграцию
(или: ./manage.py migrate someApp # только для конкретного приложения)

PS:
./manage.py migrate someApp zero #unapply все миграции для конкретного приложения

Ответ 5

Используя код @Ernest, я написал manage_custom.py для ожидающих миграции. Вы можете получить список ожидающих миграции, а также перенести эти ожидающие миграции (только), что позволит сэкономить ваше время.

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()

Применение:

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

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

PS: настройте среду согласно вашей структуре проекта.

Ответ 6

Вот мое решение Python, чтобы получить некоторую информацию о миграционных состояниях:

from io import StringIO  # for Python 2 use from StringIO import StringIO  
from django.core.management import call_command 

def get_migration_state():
    result = []
    out = StringIO()
    call_command('showmigrations', format="plan", stdout=out)
    out.seek(0)
    for line in out.readlines():
        status, name = line.rsplit(' ', 1)
        result.append((status.strip() == '[X]', name.strip()))
    return result

Результат этой функции выглядит так:

[(True, 'contenttypes.0001_initial'),
 (True, 'auth.0001_initial'),
 (False, 'admin.0001_initial'),
 (False, 'admin.0002_logentry_remove_auto_add')]

Может быть, это помогает некоторым из вас, ребята..

Ответ 7

Я проверил это, просмотрев таблицу django_migrations, в которой хранятся все примененные миграции.