如何通过代码检查Django迁移执行状态并获取迁移列表?
获取Django迁移执行状态的可行方案
方法1:通过MigrationExecutor获取官方标准状态
Django的showmigrations命令底层依赖MigrationExecutor类判断迁移状态,你可以直接复用这套逻辑,拿到已应用和待执行的迁移详情:
from django.db import connections from django.db.migrations.executor import MigrationExecutor def fetch_migration_status(): # 初始化迁移执行器(默认数据库连接) executor = MigrationExecutor(connections['default']) # 获取所有待执行的迁移计划 pending_migrations = executor.migration_plan(executor.loader.graph.leaf_nodes()) # 获取已应用的迁移集合(格式为(app_label, migration_name)元组) applied_migrations = executor.loader.applied_migrations return { "applied_count": len(applied_migrations), "pending_count": len(pending_migrations), "pending_migrations": [(mig.app_label, mig.name) for mig, _ in pending_migrations] }
这个方法和showmigrations逻辑完全一致,能准确反映系统当前迁移状态,适合需要精确监控的场景。
方法2:直接查询数据库表(轻量方案)
Django会把已应用的迁移记录存在django_migrations表中,你可以直接通过数据库查询获取已应用数量,再和所有定义的迁移总数对比:
from django.db import connections from django.db.migrations.loader import MigrationLoader def fetch_migration_counts(): loader = MigrationLoader(connections['default']) # 计算所有已定义的迁移总数 total_migrations = sum(len(migs) for migs in loader.migrated_apps.values()) # 查询已应用的迁移数量 with connections['default'].cursor() as cursor: cursor.execute("SELECT COUNT(*) FROM django_migrations") applied_count = cursor.fetchone()[0] return { "total_migrations": total_migrations, "applied_count": applied_count, "pending_count": total_migrations - applied_count }
这个方法不需要加载完整的迁移执行逻辑,性能更优,适合监控脚本这类轻量场景。
方法3:复用showmigrations命令的内部逻辑
如果需要和./manage.py showmigrations输出完全一致的结构化数据,可以直接调用该命令的内部函数:
from django.core.management.commands.showmigrations import Command as ShowMigrationsCmd from django.db import connections def fetch_detailed_migration_list(): cmd = ShowMigrationsCmd() # 禁用命令的默认输出 cmd.stdout = None cmd.style = None loader = cmd._get_loader(connections['default']) migration_status = {} # 按应用分组整理迁移状态 for app_label in loader.migrated_apps: migration_names = loader.get_migration_names(app_label) applied_set = loader.applied_migrations migration_status[app_label] = [ {"name": mig_name, "is_applied": (app_label, mig_name) in applied_set} for mig_name in migration_names ] return migration_status
注意事项
- 所有方法都需要在Django环境中运行,监控脚本需先初始化Django(设置
DJANGO_SETTINGS_MODULE环境变量,或调用django.setup())。 - 如果使用多数据库,需要替换
connections['default']为对应的数据库别名。
内容的提问来源于stack exchange,提问作者Sander Van Schoote
相关产品推荐
相关产品推荐

