检查暂停的 Django 迁移

在 Django 中,是否有一种简单的方法来检查是否已经运行了所有的数据库迁移?我找到了 manage.py migrate --list,它提供了我想要的信息,但格式不是很机器可读。

对于上下文: 我有一个脚本,在迁移数据库之前不应该开始运行。由于各种原因,从正在运行迁移的进程发送信号是很困难的。所以我想让我的脚本定期检查数据库,看看是否所有的迁移已经运行。

65594 次浏览

Shell

The only simple solution I've found so far is running

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

which will output an empty string in case all migrations have been applied.

However, it is closely tied to the output format.

Python

I checked the source code of migrate command and it seems like this should do the trick:

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

Using @Ernest code, I've written a manage_custom.py for pending migrations. You can get the list of pending migrations also migrate those pending migrations (only), hence saving your time.

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

PS: Please setup environment as per your project structure.

Try,

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

returns,

<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


Update:

If you have updated Django version >= 1.11, use below command,

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

1.10 release notes:

The new makemigrations --check option makes the command exit with a non-zero status when model changes without migrations are detected.

If you don't want to create the migrations, combine it with --dry-run:

python manage.py makemigrations --check --dry-run

Note that this doesn't check whether the migrations were applied, it only checks whether the migration files were created.

./manage.py showmigrations #check which already-made migrations have been applied or not
(or: ./manage.py showmigrations someApp #for specific app alone)

./manage.py makemigrations --dry-run #check for migrations to be made
(or: ./manage.py makemigrations someApp --dry-run #for specific app alone)

./manage.py makemigrations #make the migrations
(or: ./manage.py makemigrations someApp #for specific app alone)

./manage.py showmigrations #check which already-made migrations have been applied or not
(or: ./manage.py showmigrations someApp #for specific app alone)

./manage.py sqlmigrate someApp 0001 #view SQL changes for specific app & migration

./manage.py migrate #apply migrations
(or: ./manage.py migrate someApp #for specific app alone)

./manage.py showmigrations #check which already-made migrations have been applied or not
(or: ./manage.py showmigrations someApp #for specific app alone)

./manage.py makemigrations --dry-run #check for migrations to be made
(or: ./manage.py makemigrations someApp --dry-run #for specific app alone)

PS:
./manage.py migrate someApp zero #unapply all migrations for specific app

The documentation for these commands can be found here on the Django project's website.

Here is my Python soloution to get some information about the migration-states:

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

The result of this function looks like that:

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

Maybe it helps some of you guys..

I checked it by looking up the table django_migrations, which stores all applied migrations.

3.1 release notes

The new migrate --check option makes the command exit with a non-zero status when unapplied migrations are detected.

https://docs.djangoproject.com/en/3.1/ref/django-admin/#cmdoption-migrate-check

so finally we can

python manage.py migrate --check

Tested for Django 3.2:

python manage.py makemigrations --check --dry-run

expected output would be:

'No changes detected'

if there are no pending changes in the Models requiring a migration to be created

python manage.py migrate --plan

expected output would be:

'Planned operations: No planned migration operations.'

You can use it in python script with call_command and develop a way to check for the expected output. If there are any pending makemigrations of migrate calls, the output will be different from the expected and you can understand that something is missing.

I'm running this at a CI/CD pipeline with very good results.

Make it simple:

$ until python manage.py migrate --check ; do echo "Migration not completed" ; done