检查挂起的 Django 迁移

Posted 2023-02-24

【中文标题】检查挂起的 Django 迁移

【英文标题】：Check for pending Django migrations

【发布时间】：2015-10-28 14:31:46

【问题描述】：

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

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

【问题讨论】：

您是否正在运行自动化脚本来检查像 fabric 这样的迁移？

***.com/a/8491203/4325513

【参考方案1】：

简单点：

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

【参考方案2】：

./manage.py showmigrations #检查哪些已经进行的迁移已应用或未应用 （或：./manage.py showmigrations someApp #仅针对特定应用）

./manage.py makemigrations --dry-run #检查要进行的迁移 （或：./manage.py makemigrations someApp --dry-run #仅针对特定应用）

./manage.py makemigrations #进行迁移 （或：./manage.py makemigrations someApp #仅针对特定应用）

./manage.py showmigrations #检查哪些已经进行的迁移已应用或未应用 （或：./manage.py showmigrations someApp #仅针对特定应用）

./manage.py sqlmigrate someApp 0001 #查看特定应用程序和迁移的 SQL 更改

./manage.py migrate #apply 迁移 （或：./manage.py migrate someApp #仅针对特定应用）

./manage.py showmigrations #检查哪些已经进行的迁移已应用或未应用 （或：./manage.py showmigrations someApp #仅针对特定应用）

./manage.py makemigrations --dry-run #检查要进行的迁移 （或：./manage.py makemigrations someApp --dry-run #仅针对特定应用）

PS:./manage.py migrate someApp zero #unapply 特定应用的所有迁移

这些命令的文档可以是found here on the Django project's website。

【讨论】：

不错的答案！谢谢

致那些可能试图编辑此答案的人......此答案中的重复步骤并非偶然，如果您花时间理解问题或答案，您就会知道。如果您不检查自己（正如此答案中的重复步骤所鼓励的那样），您将永远无法掌握此过程，而且几乎肯定会破坏某些东西。如果这是一个不复杂、完全安全的过程，那么答案就不会那么长（即使没有重复）。如果没有重大的 Django 开发，我预计不会对这个答案做出任何非常有用的更改。

我唯一要添加的是link to this section of the documentation

@KoltonNoreen 添加了指向该页面的链接，但未添加该特定部分。应该做我相信的伎俩。

【参考方案3】：

针对 Django 3.2 测试：

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

预期的输出是：

'No changes detected'

如果模型中没有需要创建迁移的待处理更改

python manage.py migrate --plan

预期的输出是：

'Planned operations: No planned migration operations.'

您可以在带有 call_command 的 python 脚本中使用它，并开发一种方法来检查预期的输出。如果有任何待处理的makemigrations migrate 调用，输出将与预期不同，您可以理解缺少某些内容。

我在 CI/CD 管道上运行它，结果非常好。

【参考方案4】：

3.1 版本说明

新的 migrate --check 选项使命令在检测到未应用的迁移时以非零状态退出。

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

我们终于可以了

python manage.py migrate --check

【参考方案5】：

1.10 版本说明：

当检测到没有迁移的模型更改时，新的makemigrations --check 选项使命令以非零状态退出。

如果您不想创建迁移，请将其与 --dry-run 结合使用：

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

请注意，这不会检查是否应用了迁移，它只检查是否创建了迁移文件。

【讨论】：

虽然我认为这是对更重要问题“如何知道我是否需要进行迁移 [然后应用这些迁移]？”的答案。 OP 很可能主要或次要之后，它很容易被解释为不是对实际提出的具体问题的完全答案。我想说所有其他答案都需要一半的过程，而你的答案需要另一半。我觉得从技术上讲，OP 可能只想要一半，但实际上，对于所提出的模糊问题，两半都应该参与任何真正有用的答案。

您列出的命令与 OP 有完全相同的问题：输出不容易机器可读。我不明白你是如何解决这个问题的。 --check 命令对 OP 提出的特定问题给出了一个简单的答案，“是否运行了所有迁移”，它非常适合自动化任务，这正是它存在的原因。

我的不是作为您的直接替代品添加的。我实际上比其他人更喜欢你的答案。只是看起来不完整。 --check 看起来不错（对于它所用于的一小部分过程），尽管行为刚刚从 1.8 发生变化，这很烦人。我会再等一会儿再使用它。至于解析输出，这个过程很复杂，令人讨厌，但 grep 简单且通常不可避免。 :p 至于他的确切问题，我再次认为问题更大，没有“简单”的解决方案。如果我制作一个，将用脚本更新答案。或者其他人可以。

我希望这是正确的答案，但不幸的是它不适用于已创建但未应用的迁移。

如果您不想实际创建迁移，请添加 --dry-run 选项。

【参考方案6】：

壳牌

到目前为止我发现的唯一简单的解决方案是运行

./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

【讨论】：

在 Django 1.7 或更高版本中，您可以使用：./manage showmigrations --list 或 ./manage showmigrations --plan

【参考方案7】：

我通过查找表 django_migrations 进行了检查，该表存储了所有已应用的迁移。

【讨论】：

我建议您只检查此表以获取信息（对表的更改不适用）。我会继续使用诸如 showmigrations 之类的 django 命令，这样您就不会使数据库处于不一致的状态。

【参考方案8】：

试试，

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

【讨论】：

现在有一个内置的：./manage.py showmigrations accounts (no migrations) admin [X] 0001_initial [ ] 0002_logentry_remove_auto_add ....

【参考方案9】：

这是我的 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')]

也许它对你们中的一些人有所帮助..

【参考方案10】：

使用@Ernest 代码，我为待处理的迁移编写了manage_custom.py。您可以获取待处理迁移列表也迁移那些待处理迁移（仅），从而节省您的时间。

ma​​nage_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：请根据您的项目结构设置环境。