1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
from django.core.management.base import BaseCommand, CommandError
from django.db import DEFAULT_DB_ALIAS, connections
from django.db.migrations.loader import MigrationLoader
class Command(BaseCommand):
help = "Shows all available migrations for the current project"
def add_arguments(self, parser):
parser.add_argument('app_labels', nargs='*',
help='App labels of applications to limit the output to.')
parser.add_argument('--database', action='store', dest='database', default=DEFAULT_DB_ALIAS,
help='Nominates a database to synchronize. Defaults to the "default" database.')
formats = parser.add_mutually_exclusive_group()
formats.add_argument('--list', '-l', action='store_const', dest='format', const='list',
help='Shows a list of all migrations and which are applied.')
formats.add_argument('--plan', '-p', action='store_const', dest='format', const='plan',
help='Shows all migrations in the order they will be applied.')
parser.set_defaults(format='list')
def handle(self, *args, **options):
self.verbosity = options.get('verbosity')
# Get the database we're operating from
db = options.get('database')
connection = connections[db]
if options['format'] == "plan":
return self.show_plan(connection)
else:
return self.show_list(connection, options['app_labels'])
def show_list(self, connection, app_names=None):
"""
Shows a list of all migrations on the system, or only those of
some named apps.
"""
# Load migrations from disk/DB
loader = MigrationLoader(connection, ignore_no_migrations=True)
graph = loader.graph
# If we were passed a list of apps, validate it
if app_names:
invalid_apps = []
for app_name in app_names:
if app_name not in loader.migrated_apps:
invalid_apps.append(app_name)
if invalid_apps:
raise CommandError("No migrations present for: %s" % (", ".join(invalid_apps)))
# Otherwise, show all apps in alphabetic order
else:
app_names = sorted(loader.migrated_apps)
# For each app, print its migrations in order from oldest (roots) to
# newest (leaves).
for app_name in app_names:
self.stdout.write(app_name, self.style.MIGRATE_LABEL)
shown = set()
for node in graph.leaf_nodes(app_name):
for plan_node in graph.forwards_plan(node):
if plan_node not in shown and plan_node[0] == app_name:
# Give it a nice title if it's a squashed one
title = plan_node[1]
if graph.nodes[plan_node].replaces:
title += " (%s squashed migrations)" % len(graph.nodes[plan_node].replaces)
# Mark it as applied/unapplied
if plan_node in loader.applied_migrations:
self.stdout.write(" [X] %s" % title)
else:
self.stdout.write(" [ ] %s" % title)
shown.add(plan_node)
# If we didn't print anything, then a small message
if not shown:
self.stdout.write(" (no migrations)", self.style.MIGRATE_FAILURE)
def show_plan(self, connection):
"""
Shows all known migrations in the order they will be applied
"""
# Load migrations from disk/DB
loader = MigrationLoader(connection)
graph = loader.graph
targets = graph.leaf_nodes()
plan = []
seen = set()
# Generate the plan
for target in targets:
for migration in graph.forwards_plan(target):
if migration not in seen:
plan.append(graph.nodes[migration])
seen.add(migration)
# Output
def print_deps(migration):
out = []
for dep in migration.dependencies:
if dep[1] == "__first__":
roots = graph.root_nodes(dep[0])
dep = roots[0] if roots else (dep[0], "__first__")
out.append("%s.%s" % dep)
if out:
return " ... (%s)" % ", ".join(out)
return ""
for migration in plan:
deps = ""
if self.verbosity >= 2:
deps = print_deps(migration)
if (migration.app_label, migration.name) in loader.applied_migrations:
self.stdout.write("[X] %s%s" % (migration, deps))
else:
self.stdout.write("[ ] %s%s" % (migration, deps))
|