LoadFixtures Django migration operation

operations.py

from __future__ import unicode_literals

from os.path import splitext

from django.core import serializers
from django.db.migrations.operations.base import Operation

###############################################################################
#                                                                             #
# In order to make the following code work, the model serializers and         #
# deserializers need to be able to handle non-global app registries (i.e. get #
# `state.apps` from the migrations passed in, and not use back to             #
# `django.apps.apps`.                                                         #
#                                                                             #
# A work in progress feature can be found at                                  #
# https://github.com/MarkusH/django/compare/master...apps-aware-serializers   #
#                                                                             #
###############################################################################


class LoadFixture(Operation):

    reduces_to_sql = False
    noop = []

    def __init__(self, fixtures, reversible=True, atomic=True, hints=None):
        self.atomic = atomic
        if not isinstance(fixtures, list):
            raise ValueError("LoadFixture must be supplied with a list of fixtures")
        self.fixtures = fixtures
        self.reversible = reversible
        self.hints = hints or {}

    def deconstruct(self):
        kwargs = {
            'files': self.files,
        }
        if self.reverse_fixtures is not None:
            kwargs['reverse_fixtures'] = self.reverse_fixtures
        if self.atomic is not True:
            kwargs['atomic'] = self.atomic
        if self.hints:
            kwargs['hints'] = self.hints
        return (
            self.__class__.__name__,
            [],
            kwargs
        )

    def state_forwards(self, app_label, state):
        # LoadFixture objects have no state effect. To add some, combine this
        # with SeparateDatabaseAndState.
        pass

    def database_forwards(self, app_label, schema_editor, from_state, to_state):
        if self.allowed_to_migrate(schema_editor.connection.alias, None, hints=self.hints):
            for fixture in self.fixtures:
                format = LoadFixture.get_format(fixture)
                with open(fixture, 'rb') as fp:
                    objects = serializers.deserialize(format, fp, apps=from_state.apps)
                    for obj in objects:
                        obj.save()

    def database_backwards(self, app_label, schema_editor, from_state, to_state):
        if not self.reversible:
            raise NotImplementedError("You cannot reverse this operation")
        if self.allowed_to_migrate(schema_editor.connection.alias, None, hints=self.hints):
            for fixture in reversed(self.fixtures):
                format = LoadFixture.get_format(fixture)
                with open(fixture, 'rb') as fp:
                    objects = list(serializers.deserialize(format, fp, apps=from_state.apps))
                    for obj in reversed(objects):
                        obj.object.delete()

    def describe(self):
        return "Load Fixtures operation"

    @staticmethod
    def get_format(fixture):
        format = splitext(fixture)[1][1:]
        if not format:
            raise ValueError(
                "Problem installing fixture '%s': '%s' is not a known "
                "serialization format." % (fixture, format)
            )
        return format