Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/core/management/commands/migrate.py: 35%
215 statements
« prev ^ index » next coverage.py v6.4.4, created at 2023-07-17 14:22 -0600
1import sys
2import time
3from importlib import import_module
5from django.apps import apps
6from django.core.management.base import BaseCommand, CommandError, no_translations
7from django.core.management.sql import emit_post_migrate_signal, emit_pre_migrate_signal
8from django.db import DEFAULT_DB_ALIAS, connections, router
9from django.db.migrations.autodetector import MigrationAutodetector
10from django.db.migrations.executor import MigrationExecutor
11from django.db.migrations.loader import AmbiguityError
12from django.db.migrations.state import ModelState, ProjectState
13from django.utils.module_loading import module_has_submodule
14from django.utils.text import Truncator
17class Command(BaseCommand):
18 help = (
19 "Updates database schema. Manages both apps with migrations and those without."
20 )
21 requires_system_checks = []
23 def add_arguments(self, parser):
24 parser.add_argument(
25 "--skip-checks",
26 action="store_true",
27 help="Skip system checks.",
28 )
29 parser.add_argument(
30 "app_label",
31 nargs="?",
32 help="App label of an application to synchronize the state.",
33 )
34 parser.add_argument(
35 "migration_name",
36 nargs="?",
37 help="Database state will be brought to the state after that "
38 'migration. Use the name "zero" to unapply all migrations.',
39 )
40 parser.add_argument(
41 "--noinput",
42 "--no-input",
43 action="store_false",
44 dest="interactive",
45 help="Tells Django to NOT prompt the user for input of any kind.",
46 )
47 parser.add_argument(
48 "--database",
49 default=DEFAULT_DB_ALIAS,
50 help=(
51 'Nominates a database to synchronize. Defaults to the "default" '
52 "database."
53 ),
54 )
55 parser.add_argument(
56 "--fake",
57 action="store_true",
58 help="Mark migrations as run without actually running them.",
59 )
60 parser.add_argument(
61 "--fake-initial",
62 action="store_true",
63 help=(
64 "Detect if tables already exist and fake-apply initial migrations if "
65 "so. Make sure that the current database schema matches your initial "
66 "migration before using this flag. Django will only check for an "
67 "existing table name."
68 ),
69 )
70 parser.add_argument(
71 "--plan",
72 action="store_true",
73 help="Shows a list of the migration actions that will be performed.",
74 )
75 parser.add_argument(
76 "--run-syncdb",
77 action="store_true",
78 help="Creates tables for apps without migrations.",
79 )
80 parser.add_argument(
81 "--check",
82 action="store_true",
83 dest="check_unapplied",
84 help="Exits with a non-zero status if unapplied migrations exist.",
85 )
87 @no_translations
88 def handle(self, *args, **options):
89 database = options["database"]
90 if not options["skip_checks"]: 90 ↛ 91line 90 didn't jump to line 91, because the condition on line 90 was never true
91 self.check(databases=[database])
93 self.verbosity = options["verbosity"]
94 self.interactive = options["interactive"]
96 # Import the 'management' module within each installed app, to register
97 # dispatcher events.
98 for app_config in apps.get_app_configs():
99 if module_has_submodule(app_config.module, "management"):
100 import_module(".management", app_config.name)
102 # Get the database we're operating from
103 connection = connections[database]
105 # Hook for backends needing any database preparation
106 connection.prepare_database()
107 # Work out which apps have migrations and which do not
108 executor = MigrationExecutor(connection, self.migration_progress_callback)
110 # Raise an error if any migrations are applied before their dependencies.
111 executor.loader.check_consistent_history(connection)
113 # Before anything else, see if there's conflicting apps and drop out
114 # hard if there are any
115 conflicts = executor.loader.detect_conflicts()
116 if conflicts: 116 ↛ 117line 116 didn't jump to line 117, because the condition on line 116 was never true
117 name_str = "; ".join(
118 "%s in %s" % (", ".join(names), app) for app, names in conflicts.items()
119 )
120 raise CommandError(
121 "Conflicting migrations detected; multiple leaf nodes in the "
122 "migration graph: (%s).\nTo fix them run "
123 "'python manage.py makemigrations --merge'" % name_str
124 )
126 # If they supplied command line arguments, work out what they mean.
127 run_syncdb = options["run_syncdb"]
128 target_app_labels_only = True
129 if options["app_label"]: 129 ↛ 131line 129 didn't jump to line 131, because the condition on line 129 was never true
130 # Validate app_label.
131 app_label = options["app_label"]
132 try:
133 apps.get_app_config(app_label)
134 except LookupError as err:
135 raise CommandError(str(err))
136 if run_syncdb:
137 if app_label in executor.loader.migrated_apps:
138 raise CommandError(
139 "Can't use run_syncdb with app '%s' as it has migrations."
140 % app_label
141 )
142 elif app_label not in executor.loader.migrated_apps:
143 raise CommandError("App '%s' does not have migrations." % app_label)
145 if options["app_label"] and options["migration_name"]: 145 ↛ 146line 145 didn't jump to line 146, because the condition on line 145 was never true
146 migration_name = options["migration_name"]
147 if migration_name == "zero":
148 targets = [(app_label, None)]
149 else:
150 try:
151 migration = executor.loader.get_migration_by_prefix(
152 app_label, migration_name
153 )
154 except AmbiguityError:
155 raise CommandError(
156 "More than one migration matches '%s' in app '%s'. "
157 "Please be more specific." % (migration_name, app_label)
158 )
159 except KeyError:
160 raise CommandError(
161 "Cannot find a migration matching '%s' from app '%s'."
162 % (migration_name, app_label)
163 )
164 target = (app_label, migration.name)
165 # Partially applied squashed migrations are not included in the
166 # graph, use the last replacement instead.
167 if (
168 target not in executor.loader.graph.nodes
169 and target in executor.loader.replacements
170 ):
171 incomplete_migration = executor.loader.replacements[target]
172 target = incomplete_migration.replaces[-1]
173 targets = [target]
174 target_app_labels_only = False
175 elif options["app_label"]: 175 ↛ 176line 175 didn't jump to line 176, because the condition on line 175 was never true
176 targets = [
177 key for key in executor.loader.graph.leaf_nodes() if key[0] == app_label
178 ]
179 else:
180 targets = executor.loader.graph.leaf_nodes()
182 plan = executor.migration_plan(targets)
183 exit_dry = plan and options["check_unapplied"]
185 if options["plan"]: 185 ↛ 186line 185 didn't jump to line 186, because the condition on line 185 was never true
186 self.stdout.write("Planned operations:", self.style.MIGRATE_LABEL)
187 if not plan:
188 self.stdout.write(" No planned migration operations.")
189 for migration, backwards in plan:
190 self.stdout.write(str(migration), self.style.MIGRATE_HEADING)
191 for operation in migration.operations:
192 message, is_error = self.describe_operation(operation, backwards)
193 style = self.style.WARNING if is_error else None
194 self.stdout.write(" " + message, style)
195 if exit_dry:
196 sys.exit(1)
197 return
198 if exit_dry: 198 ↛ 199line 198 didn't jump to line 199, because the condition on line 198 was never true
199 sys.exit(1)
201 # At this point, ignore run_syncdb if there aren't any apps to sync.
202 run_syncdb = options["run_syncdb"] and executor.loader.unmigrated_apps
203 # Print some useful info
204 if self.verbosity >= 1: 204 ↛ 205line 204 didn't jump to line 205, because the condition on line 204 was never true
205 self.stdout.write(self.style.MIGRATE_HEADING("Operations to perform:"))
206 if run_syncdb:
207 if options["app_label"]:
208 self.stdout.write(
209 self.style.MIGRATE_LABEL(
210 " Synchronize unmigrated app: %s" % app_label
211 )
212 )
213 else:
214 self.stdout.write(
215 self.style.MIGRATE_LABEL(" Synchronize unmigrated apps: ")
216 + (", ".join(sorted(executor.loader.unmigrated_apps)))
217 )
218 if target_app_labels_only:
219 self.stdout.write(
220 self.style.MIGRATE_LABEL(" Apply all migrations: ")
221 + (", ".join(sorted({a for a, n in targets})) or "(none)")
222 )
223 else:
224 if targets[0][1] is None:
225 self.stdout.write(
226 self.style.MIGRATE_LABEL(" Unapply all migrations: ")
227 + str(targets[0][0])
228 )
229 else:
230 self.stdout.write(
231 self.style.MIGRATE_LABEL(" Target specific migration: ")
232 + "%s, from %s" % (targets[0][1], targets[0][0])
233 )
235 pre_migrate_state = executor._create_project_state(with_applied_migrations=True)
236 pre_migrate_apps = pre_migrate_state.apps
237 emit_pre_migrate_signal(
238 self.verbosity,
239 self.interactive,
240 connection.alias,
241 stdout=self.stdout,
242 apps=pre_migrate_apps,
243 plan=plan,
244 )
246 # Run the syncdb phase.
247 if run_syncdb: 247 ↛ 258line 247 didn't jump to line 258, because the condition on line 247 was never false
248 if self.verbosity >= 1: 248 ↛ 249line 248 didn't jump to line 249, because the condition on line 248 was never true
249 self.stdout.write(
250 self.style.MIGRATE_HEADING("Synchronizing apps without migrations:")
251 )
252 if options["app_label"]: 252 ↛ 253line 252 didn't jump to line 253, because the condition on line 252 was never true
253 self.sync_apps(connection, [app_label])
254 else:
255 self.sync_apps(connection, executor.loader.unmigrated_apps)
257 # Migrate!
258 if self.verbosity >= 1: 258 ↛ 259line 258 didn't jump to line 259, because the condition on line 258 was never true
259 self.stdout.write(self.style.MIGRATE_HEADING("Running migrations:"))
260 if not plan: 260 ↛ 261line 260 didn't jump to line 261, because the condition on line 260 was never true
261 if self.verbosity >= 1:
262 self.stdout.write(" No migrations to apply.")
263 # If there's changes that aren't in migrations yet, tell them
264 # how to fix it.
265 autodetector = MigrationAutodetector(
266 executor.loader.project_state(),
267 ProjectState.from_apps(apps),
268 )
269 changes = autodetector.changes(graph=executor.loader.graph)
270 if changes:
271 self.stdout.write(
272 self.style.NOTICE(
273 " Your models in app(s): %s have changes that are not "
274 "yet reflected in a migration, and so won't be "
275 "applied." % ", ".join(repr(app) for app in sorted(changes))
276 )
277 )
278 self.stdout.write(
279 self.style.NOTICE(
280 " Run 'manage.py makemigrations' to make new "
281 "migrations, and then re-run 'manage.py migrate' to "
282 "apply them."
283 )
284 )
285 fake = False
286 fake_initial = False
287 else:
288 fake = options["fake"]
289 fake_initial = options["fake_initial"]
290 post_migrate_state = executor.migrate(
291 targets,
292 plan=plan,
293 state=pre_migrate_state.clone(),
294 fake=fake,
295 fake_initial=fake_initial,
296 )
297 # post_migrate signals have access to all models. Ensure that all models
298 # are reloaded in case any are delayed.
299 post_migrate_state.clear_delayed_apps_cache()
300 post_migrate_apps = post_migrate_state.apps
302 # Re-render models of real apps to include relationships now that
303 # we've got a final state. This wouldn't be necessary if real apps
304 # models were rendered with relationships in the first place.
305 with post_migrate_apps.bulk_update():
306 model_keys = []
307 for model_state in post_migrate_apps.real_models: 307 ↛ 308line 307 didn't jump to line 308, because the loop on line 307 never started
308 model_key = model_state.app_label, model_state.name_lower
309 model_keys.append(model_key)
310 post_migrate_apps.unregister_model(*model_key)
311 post_migrate_apps.render_multiple(
312 [ModelState.from_model(apps.get_model(*model)) for model in model_keys]
313 )
315 # Send the post_migrate signal, so individual apps can do whatever they need
316 # to do at this point.
317 emit_post_migrate_signal(
318 self.verbosity,
319 self.interactive,
320 connection.alias,
321 stdout=self.stdout,
322 apps=post_migrate_apps,
323 plan=plan,
324 )
326 def migration_progress_callback(self, action, migration=None, fake=False):
327 if self.verbosity >= 1: 327 ↛ 328line 327 didn't jump to line 328, because the condition on line 327 was never true
328 compute_time = self.verbosity > 1
329 if action == "apply_start":
330 if compute_time:
331 self.start = time.monotonic()
332 self.stdout.write(" Applying %s..." % migration, ending="")
333 self.stdout.flush()
334 elif action == "apply_success":
335 elapsed = (
336 " (%.3fs)" % (time.monotonic() - self.start) if compute_time else ""
337 )
338 if fake:
339 self.stdout.write(self.style.SUCCESS(" FAKED" + elapsed))
340 else:
341 self.stdout.write(self.style.SUCCESS(" OK" + elapsed))
342 elif action == "unapply_start":
343 if compute_time:
344 self.start = time.monotonic()
345 self.stdout.write(" Unapplying %s..." % migration, ending="")
346 self.stdout.flush()
347 elif action == "unapply_success":
348 elapsed = (
349 " (%.3fs)" % (time.monotonic() - self.start) if compute_time else ""
350 )
351 if fake:
352 self.stdout.write(self.style.SUCCESS(" FAKED" + elapsed))
353 else:
354 self.stdout.write(self.style.SUCCESS(" OK" + elapsed))
355 elif action == "render_start":
356 if compute_time:
357 self.start = time.monotonic()
358 self.stdout.write(" Rendering model states...", ending="")
359 self.stdout.flush()
360 elif action == "render_success":
361 elapsed = (
362 " (%.3fs)" % (time.monotonic() - self.start) if compute_time else ""
363 )
364 self.stdout.write(self.style.SUCCESS(" DONE" + elapsed))
366 def sync_apps(self, connection, app_labels):
367 """Run the old syncdb-style operation on a list of app_labels."""
368 with connection.cursor() as cursor:
369 tables = connection.introspection.table_names(cursor)
371 # Build the manifest of apps and models that are to be synchronized.
372 all_models = [
373 (
374 app_config.label,
375 router.get_migratable_models(
376 app_config, connection.alias, include_auto_created=False
377 ),
378 )
379 for app_config in apps.get_app_configs()
380 if app_config.models_module is not None and app_config.label in app_labels
381 ]
383 def model_installed(model):
384 opts = model._meta
385 converter = connection.introspection.identifier_converter
386 return not (
387 (converter(opts.db_table) in tables)
388 or (
389 opts.auto_created
390 and converter(opts.auto_created._meta.db_table) in tables
391 )
392 )
394 manifest = {
395 app_name: list(filter(model_installed, model_list))
396 for app_name, model_list in all_models
397 }
399 # Create the tables for each model
400 if self.verbosity >= 1: 400 ↛ 401line 400 didn't jump to line 401, because the condition on line 400 was never true
401 self.stdout.write(" Creating tables...")
402 with connection.schema_editor() as editor:
403 for app_name, model_list in manifest.items():
404 for model in model_list: 404 ↛ 406line 404 didn't jump to line 406, because the loop on line 404 never started
405 # Never install unmanaged models, etc.
406 if not model._meta.can_migrate(connection):
407 continue
408 if self.verbosity >= 3:
409 self.stdout.write(
410 " Processing %s.%s model"
411 % (app_name, model._meta.object_name)
412 )
413 if self.verbosity >= 1:
414 self.stdout.write(
415 " Creating table %s" % model._meta.db_table
416 )
417 editor.create_model(model)
419 # Deferred SQL is executed when exiting the editor's context.
420 if self.verbosity >= 1: 420 ↛ 421line 420 didn't jump to line 421, because the condition on line 420 was never true
421 self.stdout.write(" Running deferred SQL...")
423 @staticmethod
424 def describe_operation(operation, backwards):
425 """Return a string that describes a migration operation for --plan."""
426 prefix = ""
427 is_error = False
428 if hasattr(operation, "code"):
429 code = operation.reverse_code if backwards else operation.code
430 action = (code.__doc__ or "") if code else None
431 elif hasattr(operation, "sql"):
432 action = operation.reverse_sql if backwards else operation.sql
433 else:
434 action = ""
435 if backwards:
436 prefix = "Undo "
437 if action is not None:
438 action = str(action).replace("\n", "")
439 elif backwards:
440 action = "IRREVERSIBLE"
441 is_error = True
442 if action:
443 action = " -> " + action
444 truncated = Truncator(action)
445 return prefix + operation.describe() + truncated.chars(40), is_error