384 lines
16 KiB
Python
384 lines
16 KiB
Python
from __future__ import unicode_literals
|
|
|
|
import datetime
|
|
import inspect
|
|
import decimal
|
|
import collections
|
|
from importlib import import_module
|
|
import os
|
|
import re
|
|
import sys
|
|
import types
|
|
|
|
from django.apps import apps
|
|
from django.db import models, migrations
|
|
from django.db.migrations.loader import MigrationLoader
|
|
from django.utils import datetime_safe, six
|
|
from django.utils.encoding import force_text
|
|
from django.utils.functional import Promise
|
|
|
|
|
|
COMPILED_REGEX_TYPE = type(re.compile(''))
|
|
|
|
|
|
class SettingsReference(str):
|
|
"""
|
|
Special subclass of string which actually references a current settings
|
|
value. It's treated as the value in memory, but serializes out to a
|
|
settings.NAME attribute reference.
|
|
"""
|
|
|
|
def __new__(self, value, setting_name):
|
|
return str.__new__(self, value)
|
|
|
|
def __init__(self, value, setting_name):
|
|
self.setting_name = setting_name
|
|
|
|
|
|
class OperationWriter(object):
|
|
indentation = 2
|
|
|
|
def __init__(self, operation):
|
|
self.operation = operation
|
|
self.buff = []
|
|
|
|
def serialize(self):
|
|
imports = set()
|
|
name, args, kwargs = self.operation.deconstruct()
|
|
argspec = inspect.getargspec(self.operation.__init__)
|
|
normalized_kwargs = inspect.getcallargs(self.operation.__init__, *args, **kwargs)
|
|
|
|
# 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 arg_name in argspec.args[1:]:
|
|
arg_value = normalized_kwargs[arg_name]
|
|
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)
|
|
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)
|
|
self.feed('%s,' % arg_string)
|
|
imports.update(arg_imports)
|
|
self.unindent()
|
|
self.feed('],')
|
|
else:
|
|
arg_string, arg_imports = MigrationWriter.serialize(arg_value)
|
|
self.feed('%s=%s,' % (arg_name, arg_string))
|
|
imports.update(arg_imports)
|
|
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(object):
|
|
"""
|
|
Takes a Migration instance and is able to produce the contents
|
|
of the migration file from it.
|
|
"""
|
|
|
|
def __init__(self, migration):
|
|
self.migration = migration
|
|
|
|
def as_string(self):
|
|
"""
|
|
Returns a string of the file contents.
|
|
"""
|
|
items = {
|
|
"replaces_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:
|
|
# No need to output bytestrings for dependencies
|
|
dependency = tuple([force_text(s) for s in dependency])
|
|
dependencies.append(" %s," % self.serialize(dependency)[0])
|
|
items["dependencies"] = "\n".join(dependencies) + "\n" if dependencies else ""
|
|
|
|
# Format imports nicely
|
|
imports.discard("from django.db import models")
|
|
items["imports"] = "\n".join(imports) + "\n" if imports else ""
|
|
|
|
# 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]
|
|
|
|
return (MIGRATION_TEMPLATE % items).encode("utf8")
|
|
|
|
@property
|
|
def filename(self):
|
|
return "%s.py" % self.migration.name
|
|
|
|
@property
|
|
def path(self):
|
|
migrations_package_name = MigrationLoader.migrations_module(self.migration.app_label)
|
|
# See if we can import the migrations module directly
|
|
try:
|
|
migrations_module = import_module(migrations_package_name)
|
|
|
|
# Python 3 fails when the migrations directory does not have a
|
|
# __init__.py file
|
|
if not hasattr(migrations_module, '__file__'):
|
|
raise ImportError
|
|
|
|
basedir = os.path.dirname(migrations_module.__file__)
|
|
except ImportError:
|
|
app_config = apps.get_app_config(self.migration.app_label)
|
|
migrations_package_basename = migrations_package_name.split(".")[-1]
|
|
|
|
# Alright, see if it's a direct submodule of the app
|
|
if '%s.%s' % (app_config.name, migrations_package_basename) == migrations_package_name:
|
|
basedir = os.path.join(app_config.path, migrations_package_basename)
|
|
else:
|
|
# In case of using MIGRATION_MODULES setting and the custom
|
|
# package doesn't exist, create one.
|
|
package_dirs = migrations_package_name.split(".")
|
|
create_path = os.path.join(sys.path[0], *package_dirs)
|
|
if not os.path.isdir(create_path):
|
|
os.makedirs(create_path)
|
|
for i in range(1, len(package_dirs) + 1):
|
|
init_dir = os.path.join(sys.path[0], *package_dirs[:i])
|
|
init_path = os.path.join(init_dir, "__init__.py")
|
|
if not os.path.isfile(init_path):
|
|
open(init_path, "w").close()
|
|
return os.path.join(create_path, self.filename)
|
|
return os.path.join(basedir, self.filename)
|
|
|
|
@classmethod
|
|
def serialize_deconstructed(cls, path, args, kwargs):
|
|
module, name = path.rsplit(".", 1)
|
|
if module == "django.db.models":
|
|
imports = set(["from django.db import models"])
|
|
name = "models.%s" % name
|
|
else:
|
|
imports = set(["import %s" % module])
|
|
name = path
|
|
strings = []
|
|
for arg in args:
|
|
arg_string, arg_imports = cls.serialize(arg)
|
|
strings.append(arg_string)
|
|
imports.update(arg_imports)
|
|
for kw, arg in kwargs.items():
|
|
arg_string, arg_imports = cls.serialize(arg)
|
|
imports.update(arg_imports)
|
|
strings.append("%s=%s" % (kw, arg_string))
|
|
return "%s(%s)" % (name, ", ".join(strings)), imports
|
|
|
|
@classmethod
|
|
def serialize(cls, value):
|
|
"""
|
|
Serializes the value to a string that's parsable by Python, along
|
|
with any needed imports to make that string work.
|
|
More advanced than repr() as it can encode things
|
|
like datetime.datetime.now.
|
|
"""
|
|
# FIXME: Ideally Promise would be reconstructible, but for now we
|
|
# use force_text on them and defer to the normal string serialization
|
|
# process.
|
|
if isinstance(value, Promise):
|
|
value = force_text(value)
|
|
|
|
# Sequences
|
|
if isinstance(value, (list, set, tuple)):
|
|
imports = set()
|
|
strings = []
|
|
for item in value:
|
|
item_string, item_imports = cls.serialize(item)
|
|
imports.update(item_imports)
|
|
strings.append(item_string)
|
|
if isinstance(value, set):
|
|
format = "set([%s])"
|
|
elif isinstance(value, tuple):
|
|
# When len(value)==0, the empty tuple should be serialized as
|
|
# "()", not "(,)" because (,) is invalid Python syntax.
|
|
format = "(%s)" if len(value) != 1 else "(%s,)"
|
|
else:
|
|
format = "[%s]"
|
|
return format % (", ".join(strings)), imports
|
|
# Dictionaries
|
|
elif isinstance(value, dict):
|
|
imports = set()
|
|
strings = []
|
|
for k, v in value.items():
|
|
k_string, k_imports = cls.serialize(k)
|
|
v_string, v_imports = cls.serialize(v)
|
|
imports.update(k_imports)
|
|
imports.update(v_imports)
|
|
strings.append((k_string, v_string))
|
|
return "{%s}" % (", ".join("%s: %s" % (k, v) for k, v in strings)), imports
|
|
# Datetimes
|
|
elif isinstance(value, datetime.datetime):
|
|
if value.tzinfo is not None:
|
|
raise ValueError("Cannot serialize datetime values with timezones. Either use a callable value for default or remove the timezone.")
|
|
value_repr = repr(value)
|
|
if isinstance(value, datetime_safe.datetime):
|
|
value_repr = "datetime.%s" % value_repr
|
|
return value_repr, set(["import datetime"])
|
|
# Dates
|
|
elif isinstance(value, datetime.date):
|
|
value_repr = repr(value)
|
|
if isinstance(value, datetime_safe.date):
|
|
value_repr = "datetime.%s" % value_repr
|
|
return value_repr, set(["import datetime"])
|
|
# Settings references
|
|
elif isinstance(value, SettingsReference):
|
|
return "settings.%s" % value.setting_name, set(["from django.conf import settings"])
|
|
# Simple types
|
|
elif isinstance(value, six.integer_types + (float, bool, type(None))):
|
|
return repr(value), set()
|
|
elif isinstance(value, six.binary_type):
|
|
value_repr = repr(value)
|
|
if six.PY2:
|
|
# Prepend the `b` prefix since we're importing unicode_literals
|
|
value_repr = 'b' + value_repr
|
|
return value_repr, set()
|
|
elif isinstance(value, six.text_type):
|
|
value_repr = repr(value)
|
|
if six.PY2:
|
|
# Strip the `u` prefix since we're importing unicode_literals
|
|
value_repr = value_repr[1:]
|
|
return value_repr, set()
|
|
# Decimal
|
|
elif isinstance(value, decimal.Decimal):
|
|
return repr(value), set(["from decimal import Decimal"])
|
|
# Django fields
|
|
elif isinstance(value, models.Field):
|
|
attr_name, path, args, kwargs = value.deconstruct()
|
|
return cls.serialize_deconstructed(path, args, kwargs)
|
|
# Anything that knows how to deconstruct itself.
|
|
elif hasattr(value, 'deconstruct'):
|
|
return cls.serialize_deconstructed(*value.deconstruct())
|
|
# Functions
|
|
elif isinstance(value, (types.FunctionType, types.BuiltinFunctionType)):
|
|
# @classmethod?
|
|
if getattr(value, "__self__", None) and isinstance(value.__self__, type):
|
|
klass = value.__self__
|
|
module = klass.__module__
|
|
return "%s.%s.%s" % (module, klass.__name__, value.__name__), set(["import %s" % module])
|
|
# Further error checking
|
|
if value.__name__ == '<lambda>':
|
|
raise ValueError("Cannot serialize function: lambda")
|
|
if value.__module__ is None:
|
|
raise ValueError("Cannot serialize function %r: No module" % value)
|
|
# Python 3 is a lot easier, and only uses this branch if it's not local.
|
|
if getattr(value, "__qualname__", None) and getattr(value, "__module__", None):
|
|
if "<" not in value.__qualname__: # Qualname can include <locals>
|
|
return "%s.%s" % (value.__module__, value.__qualname__), set(["import %s" % value.__module__])
|
|
# Python 2/fallback version
|
|
module_name = value.__module__
|
|
# Make sure it's actually there and not an unbound method
|
|
module = import_module(module_name)
|
|
if not hasattr(module, value.__name__):
|
|
raise ValueError(
|
|
"Could not find function %s in %s.\nPlease note that "
|
|
"due to Python 2 limitations, you cannot serialize "
|
|
"unbound method functions (e.g. a method declared\n"
|
|
"and used in the same class body). Please move the "
|
|
"function into the main module body to use migrations.\n"
|
|
"For more information, see https://docs.djangoproject.com/en/1.7/topics/migrations/#serializing-values"
|
|
% (value.__name__, module_name))
|
|
return "%s.%s" % (module_name, value.__name__), set(["import %s" % module_name])
|
|
# Classes
|
|
elif isinstance(value, type):
|
|
special_cases = [
|
|
(models.Model, "models.Model", []),
|
|
]
|
|
for case, string, imports in special_cases:
|
|
if case is value:
|
|
return string, set(imports)
|
|
if hasattr(value, "__module__"):
|
|
module = value.__module__
|
|
return "%s.%s" % (module, value.__name__), set(["import %s" % module])
|
|
# Other iterables
|
|
elif isinstance(value, collections.Iterable):
|
|
imports = set()
|
|
strings = []
|
|
for item in value:
|
|
item_string, item_imports = cls.serialize(item)
|
|
imports.update(item_imports)
|
|
strings.append(item_string)
|
|
# When len(strings)==0, the empty iterable should be serialized as
|
|
# "()", not "(,)" because (,) is invalid Python syntax.
|
|
format = "(%s)" if len(strings) != 1 else "(%s,)"
|
|
return format % (", ".join(strings)), imports
|
|
# Compiled regex
|
|
elif isinstance(value, COMPILED_REGEX_TYPE):
|
|
imports = set(["import re"])
|
|
regex_pattern, pattern_imports = cls.serialize(value.pattern)
|
|
regex_flags, flag_imports = cls.serialize(value.flags)
|
|
imports.update(pattern_imports)
|
|
imports.update(flag_imports)
|
|
args = [regex_pattern]
|
|
if value.flags:
|
|
args.append(regex_flags)
|
|
return "re.compile(%s)" % ', '.join(args), imports
|
|
# Uh oh.
|
|
else:
|
|
raise ValueError("Cannot serialize: %r\nThere are some values Django cannot serialize into migration files.\nFor more, see https://docs.djangoproject.com/en/dev/topics/migrations/#migration-serializing" % value)
|
|
|
|
|
|
MIGRATION_TEMPLATE = """\
|
|
# -*- coding: utf-8 -*-
|
|
from __future__ import unicode_literals
|
|
|
|
from django.db import models, migrations
|
|
%(imports)s
|
|
|
|
class Migration(migrations.Migration):
|
|
%(replaces_str)s
|
|
dependencies = [
|
|
%(dependencies)s\
|
|
]
|
|
|
|
operations = [
|
|
%(operations)s\
|
|
]
|
|
"""
|