post_migrate 信号 - 在所有迁移之后只调用一次答案

【问题标题】：post_migrate signal - call just once after all migrationspost_migrate 信号 - 在所有迁移之后只调用一次
【发布时间】：2023-02-06 08:38:48
【问题描述】：

在这个（非常简化的）示例中，我需要在所有应用程序迁移后做一件事（调用do_something_after_all_apps_migrated()）。

问题是 post_migrate 信号为每个应用程序调用一次。

我有一个应用程序：after_migrations

应用程序.py

from django.apps import AppConfig
from django.db.models.signals import post_migrate
from django.dispatch import receiver


class AfterMigrationsConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'after_migrations'

    def ready(self):
        @receiver(post_migrate)
        def _post_migrate(sender,**kwargs):
            app_name = kwargs['app_config'].name
            if app_name == 'after_migrations':
                do_something_after_all_apps_migrated()

如您所见，这可行，但问题是不能保证我的应用程序 (after_migrations) 是迁移的最新应用程序。这意味着不保证数据库结构。

一切迁移完成后，如何调用do_something_after_all_apps_migrated()？它不一定是一个信号。

【问题讨论】：

标签： python django django-migrations django-signals

【解决方案1】：

我不得不覆盖 migrate 命令并使用信号来实现这一点：

# app/signals.py

from django import dispatch


migrate_finished = dispatch.Signal()

# app/management/commands/migrate.py

from django.core.management.base import no_translations
from django.core.management.commands.migrate import Command as MigrateCommand

from ...signals import migrate_finished


class Command(MigrateCommand):
    @no_translations
    def handle(self, *args, **options):
        super().handle(*args, **options)
        migrate_finished.send(self)

有了它，您可以收听 migrate_finished 信号以触发您的迁移后代码。

您还可以在发送信号之前或使用此功能处理信号之前检查是否应用了所有迁移：

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


def is_database_synchronized():
    # The code is from django/core/management/commands/migrate.py
    connection = connections[DEFAULT_DB_ALIAS]
    connection.prepare_database()
    executor = MigrationExecutor(connection)
    targets = executor.loader.graph.leaf_nodes()
    return not executor.migration_plan(targets)

【讨论】：