|
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311 |
- import os
- import re
- from importlib import import_module
-
- from django import get_version
- from django.apps import apps
-
- # SettingsReference imported for backwards compatibility in Django 2.2.
- from django.conf import SettingsReference # NOQA
- from django.db import migrations
- from django.db.migrations.loader import MigrationLoader
- from django.db.migrations.serializer import Serializer, serializer_factory
- from django.utils.inspect import get_func_args
- from django.utils.module_loading import module_dir
- from django.utils.timezone import now
-
-
- class OperationWriter:
- def __init__(self, operation, indentation=2):
- self.operation = operation
- self.buff = []
- self.indentation = indentation
-
- def serialize(self):
- def _write(_arg_name, _arg_value):
- if _arg_name in self.operation.serialization_expand_args and isinstance(
- _arg_value, (list, tuple, dict)
- ):
- if isinstance(_arg_value, dict):
- self.feed("%s={" % _arg_name)
- self.indent()
- for key, value in _arg_value.items():
- key_string, key_imports = MigrationWriter.serialize(key)
- arg_string, arg_imports = MigrationWriter.serialize(value)
- args = arg_string.splitlines()
- if len(args) > 1:
- self.feed("%s: %s" % (key_string, args[0]))
- for arg in args[1:-1]:
- self.feed(arg)
- self.feed("%s," % args[-1])
- else:
- self.feed("%s: %s," % (key_string, arg_string))
- imports.update(key_imports)
- imports.update(arg_imports)
- self.unindent()
- self.feed("},")
- else:
- self.feed("%s=[" % _arg_name)
- self.indent()
- for item in _arg_value:
- arg_string, arg_imports = MigrationWriter.serialize(item)
- args = arg_string.splitlines()
- if len(args) > 1:
- for arg in args[:-1]:
- self.feed(arg)
- self.feed("%s," % args[-1])
- else:
- self.feed("%s," % arg_string)
- imports.update(arg_imports)
- self.unindent()
- self.feed("],")
- else:
- arg_string, arg_imports = MigrationWriter.serialize(_arg_value)
- args = arg_string.splitlines()
- if len(args) > 1:
- self.feed("%s=%s" % (_arg_name, args[0]))
- for arg in args[1:-1]:
- self.feed(arg)
- self.feed("%s," % args[-1])
- else:
- self.feed("%s=%s," % (_arg_name, arg_string))
- imports.update(arg_imports)
-
- imports = set()
- name, args, kwargs = self.operation.deconstruct()
- operation_args = get_func_args(self.operation.__init__)
-
- # See if this operation is in django.db.migrations. If it is,
- # We can just use the fact we already have that imported,
- # otherwise, we need to add an import for the operation class.
- if getattr(migrations, name, None) == self.operation.__class__:
- self.feed("migrations.%s(" % name)
- else:
- imports.add("import %s" % (self.operation.__class__.__module__))
- self.feed("%s.%s(" % (self.operation.__class__.__module__, name))
-
- self.indent()
-
- for i, arg in enumerate(args):
- arg_value = arg
- arg_name = operation_args[i]
- _write(arg_name, arg_value)
-
- i = len(args)
- # Only iterate over remaining arguments
- for arg_name in operation_args[i:]:
- if arg_name in kwargs: # Don't sort to maintain signature order
- arg_value = kwargs[arg_name]
- _write(arg_name, arg_value)
-
- self.unindent()
- self.feed("),")
- return self.render(), imports
-
- def indent(self):
- self.indentation += 1
-
- def unindent(self):
- self.indentation -= 1
-
- def feed(self, line):
- self.buff.append(" " * (self.indentation * 4) + line)
-
- def render(self):
- return "\n".join(self.buff)
-
-
- class MigrationWriter:
- """
- Take a Migration instance and is able to produce the contents
- of the migration file from it.
- """
-
- def __init__(self, migration, include_header=True):
- self.migration = migration
- self.include_header = include_header
- self.needs_manual_porting = False
-
- def as_string(self):
- """Return a string of the file contents."""
- items = {
- "replaces_str": "",
- "initial_str": "",
- }
-
- imports = set()
-
- # Deconstruct operations
- operations = []
- for operation in self.migration.operations:
- operation_string, operation_imports = OperationWriter(operation).serialize()
- imports.update(operation_imports)
- operations.append(operation_string)
- items["operations"] = "\n".join(operations) + "\n" if operations else ""
-
- # Format dependencies and write out swappable dependencies right
- dependencies = []
- for dependency in self.migration.dependencies:
- if dependency[0] == "__setting__":
- dependencies.append(
- " migrations.swappable_dependency(settings.%s),"
- % dependency[1]
- )
- imports.add("from django.conf import settings")
- else:
- dependencies.append(" %s," % self.serialize(dependency)[0])
- items["dependencies"] = "\n".join(dependencies) + "\n" if dependencies else ""
-
- # Format imports nicely, swapping imports of functions from migration files
- # for comments
- migration_imports = set()
- for line in list(imports):
- if re.match(r"^import (.*)\.\d+[^\s]*$", line):
- migration_imports.add(line.split("import")[1].strip())
- imports.remove(line)
- self.needs_manual_porting = True
-
- # django.db.migrations is always used, but models import may not be.
- # If models import exists, merge it with migrations import.
- if "from django.db import models" in imports:
- imports.discard("from django.db import models")
- imports.add("from django.db import migrations, models")
- else:
- imports.add("from django.db import migrations")
-
- # Sort imports by the package / module to be imported (the part after
- # "from" in "from ... import ..." or after "import" in "import ...").
- sorted_imports = sorted(imports, key=lambda i: i.split()[1])
- items["imports"] = "\n".join(sorted_imports) + "\n" if imports else ""
- if migration_imports:
- items["imports"] += (
- "\n\n# Functions from the following migrations need manual "
- "copying.\n# Move them and any dependencies into this file, "
- "then update the\n# RunPython operations to refer to the local "
- "versions:\n# %s"
- ) % "\n# ".join(sorted(migration_imports))
- # If there's a replaces, make a string for it
- if self.migration.replaces:
- items["replaces_str"] = (
- "\n replaces = %s\n" % self.serialize(self.migration.replaces)[0]
- )
- # Hinting that goes into comment
- if self.include_header:
- items["migration_header"] = MIGRATION_HEADER_TEMPLATE % {
- "version": get_version(),
- "timestamp": now().strftime("%Y-%m-%d %H:%M"),
- }
- else:
- items["migration_header"] = ""
-
- if self.migration.initial:
- items["initial_str"] = "\n initial = True\n"
-
- return MIGRATION_TEMPLATE % items
-
- @property
- def basedir(self):
- migrations_package_name, _ = MigrationLoader.migrations_module(
- self.migration.app_label
- )
-
- if migrations_package_name is None:
- raise ValueError(
- "Django can't create migrations for app '%s' because "
- "migrations have been disabled via the MIGRATION_MODULES "
- "setting." % self.migration.app_label
- )
-
- # See if we can import the migrations module directly
- try:
- migrations_module = import_module(migrations_package_name)
- except ImportError:
- pass
- else:
- try:
- return module_dir(migrations_module)
- except ValueError:
- pass
-
- # Alright, see if it's a direct submodule of the app
- app_config = apps.get_app_config(self.migration.app_label)
- (
- maybe_app_name,
- _,
- migrations_package_basename,
- ) = migrations_package_name.rpartition(".")
- if app_config.name == maybe_app_name:
- return os.path.join(app_config.path, migrations_package_basename)
-
- # In case of using MIGRATION_MODULES setting and the custom package
- # doesn't exist, create one, starting from an existing package
- existing_dirs, missing_dirs = migrations_package_name.split("."), []
- while existing_dirs:
- missing_dirs.insert(0, existing_dirs.pop(-1))
- try:
- base_module = import_module(".".join(existing_dirs))
- except (ImportError, ValueError):
- continue
- else:
- try:
- base_dir = module_dir(base_module)
- except ValueError:
- continue
- else:
- break
- else:
- raise ValueError(
- "Could not locate an appropriate location to create "
- "migrations package %s. Make sure the toplevel "
- "package exists and can be imported." % migrations_package_name
- )
-
- final_dir = os.path.join(base_dir, *missing_dirs)
- os.makedirs(final_dir, exist_ok=True)
- for missing_dir in missing_dirs:
- base_dir = os.path.join(base_dir, missing_dir)
- with open(os.path.join(base_dir, "__init__.py"), "w"):
- pass
-
- return final_dir
-
- @property
- def filename(self):
- return "%s.py" % self.migration.name
-
- @property
- def path(self):
- return os.path.join(self.basedir, self.filename)
-
- @classmethod
- def serialize(cls, value):
- return serializer_factory(value).serialize()
-
- @classmethod
- def register_serializer(cls, type_, serializer):
- Serializer.register(type_, serializer)
-
- @classmethod
- def unregister_serializer(cls, type_):
- Serializer.unregister(type_)
-
-
- MIGRATION_HEADER_TEMPLATE = """\
- # Generated by Django %(version)s on %(timestamp)s
-
- """
-
-
- MIGRATION_TEMPLATE = """\
- %(migration_header)s%(imports)s
-
- class Migration(migrations.Migration):
- %(replaces_str)s%(initial_str)s
- dependencies = [
- %(dependencies)s\
- ]
-
- operations = [
- %(operations)s\
- ]
- """
|