Python django.core.serializers 模块，deserialize() 实例源码

def load_fixture(apps, schema_editor):
    from django.conf import settings
    if settings.TESTS_RUNNING:
        # Currently we don't have any test relying to borders, we skip
        # to accelerate tests
        return

    fixture_file = os.path.join(fixture_dir, fixture_filename)

    with open(fixture_file) as fixture:
        objects = serializers.deserialize('json', fixture, ignorenonexistent=True)
        for obj in objects:
            obj.save()

# Most probably we won't add some revertion code to go to 0001, so this code
# won't execute ever and I take it out for coverage. At some point we may
# delete. Stefanos 2015-10-12 16:46:00+03:00

def test_serialization_partial(self):
        o1 = models.PartialVersionModel(
            partial=Version('0.1.1', partial=True),
            optional=Version('0.2.4-rc42', partial=True),
            optional_spec=None,
        )
        o2 = models.PartialVersionModel(
            partial=Version('0.4.3-rc3+build3', partial=True),
            optional='',
            optional_spec=Spec('==0.1.1,!=0.1.1-alpha'),
        )

        data = serializers.serialize('json', [o1, o2])

        obj1, obj2 = serializers.deserialize('json', data)
        self.assertEqual(o1.partial, obj1.object.partial)
        self.assertEqual(o1.optional, obj1.object.optional)
        self.assertEqual(o2.partial, obj2.object.partial)
        self.assertEqual(o2.optional, obj2.object.optional)

def setUp(self):
        self.image = Image.objects.create(
            title="Test image",
            file=get_test_image_file(),
        )

        # Create an image with a missing file, by deserializing fom a python object
        # (which bypasses FileField's attempt to read the file)
        self.bad_image = list(serializers.deserialize('python', [{
            'fields': {
                'title': 'missing image',
                'height': 100,
                'file': 'original_images/missing-image.jpg',
                'width': 100,
            },
            'model': 'wagtailimages.image'
        }]))[0].object
        self.bad_image.save()

def setUp(self):
        self.engine = engines['jinja2']

        self.image = Image.objects.create(
            title="Test image",
            file=get_test_image_file(),
        )

        # Create an image with a missing file, by deserializing fom a python object
        # (which bypasses FileField's attempt to read the file)
        self.bad_image = list(serializers.deserialize('python', [{
            'fields': {
                'title': 'missing image',
                'height': 100,
                'file': 'original_images/missing-image.jpg',
                'width': 100,
            },
            'model': 'wagtailimages.image'
        }]))[0].object
        self.bad_image.save()

def testSerialization(self):
        model = MinimalTestingModel(pickle_field={'foo': 'bar'})
        serialized = serializers.serialize('json', [model])
        data = json.loads(serialized)

        # determine output at runtime, because pickle output in python 3
        # is different (but compatible with python 2)
        p = dbsafe_encode({'foo': 'bar'})

        self.assertEquals(data,
            [{'pk': None, 'model': 'picklefield.minimaltestingmodel',
              'fields': {"pickle_field": p}}])

        for deserialized_test in serializers.deserialize('json', serialized):
            self.assertEquals(deserialized_test.object,
                              model)

def test_model_deserialization_json(self):
        dump = (
            '[{'
            '"fields": {"values": "|3|1|2|"}, '
            '"model": "tests.integertuplemodel", "pk": 1'
            '}]'
        )
        obj = list(deserialize('json', dump))[0].object
        self.assertEqual(obj.values, (3, 1, 2))

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize_db_from_string(self, data):
        """
        Reloads the database with data from a string generated by
        the serialize_db_to_string method.
        """
        data = StringIO(data)
        for obj in serializers.deserialize("json", data, using=self.connection.alias):
            obj.save()

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def test_deserialize(self):
        pk = Car.objects.latest("id").id + 1
        json_car = json.dumps([{"model": "readonly_app.car",
                                "pk": pk,
                                "fields": {"wheel_number": 12,
                                           "manufacturer": "Volvo"}}])

        # Check that the save works
        with self.assertSQLQueries(Car) as capture:
            deserialized = serializers.deserialize("json", json_car)

            car, = deserialized
            car.save()

        # Because a pk is specified, Django will try to do an UPDATE and then
        # an INSERT when the UPDATE returns with 0 rows affected
        self.assertNumCommands(capture, command="UPDATE", count=1)
        self.assertNumCommands(capture, command="INSERT", count=1)

        car = Car.objects.get(pk=pk)
        self.assertEqual(car.wheel_number, 12)
        self.assertEqual(car.manufacturer, "Renault")

        json_car = json_car.replace("12", "14")
        # Check that the UPDATE works
        with self.assertSQLQueries(Car) as capture:
            deserialized = serializers.deserialize("json", json_car)

            car, = deserialized
            car.save()

        self.assertNumCommands(capture, command="UPDATE", count=1)
        self.assertNumCommands(capture, command="INSERT", count=0)

        car = Car.objects.get(pk=pk)
        self.assertEqual(car.wheel_number, 14)
        self.assertEqual(car.manufacturer, "Renault")

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def field_value(self):
        instances = serializers.deserialize('json', self.serialized_data)
        instance = list(instances)[0].object
        return getattr(instance, self.field_name)

def load_fixture(cls, fixture_spec, apps, schema_editor):
        app_name, model_name, fixture_file_name = fixture_spec
        original_apps = serializers.python.apps
        serializers.python.apps = apps
        queryset = cls.get_model_class(app_name, model_name,
                                       apps, schema_editor)
        path, fixture_type = cls.fixture_path(app_name, fixture_file_name)
        fixture_file = open(path)
        objects = serializers.deserialize(fixture_type,
                                          fixture_file,
                                          ignorenonexistent=True)
        for obj in objects:
            cls.update_or_create_object(obj, queryset)
        fixture_file.close()
        serializers.python.apps = original_apps

def test_integer_loading(self):
        instance = list(
            serializers.deserialize('json', self.test_data_integer))[0].object
        self.assertEqual(42, instance.field)

def test_char_loading(self):
        instance = list(serializers.deserialize('json',
                                                self.test_data_char))[0].object
        self.assertEqual('Hello, world!', instance.field)

def test_loading(self):
        instance = list(serializers.deserialize('json',
                                                self.test_data))[0].object
        self.assertEqual([1, 2, None], instance.field)

def load_fixture(apps, schema_editor):
    fixture_file = os.path.join(fixture_dir, fixture_filename)

    fixture = open(fixture_file, 'rb')
    objects = serializers.deserialize('json', fixture, ignorenonexistent=True)
    for obj in objects:
        obj.save()
    fixture.close()

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def test_serializer(self, format='json'):
        from django.core import serializers
        created = datetime.now()
        x = SerializeModel(key_name='blue_key', name='blue', count=4)
        x.put()
        SerializeModel(name='green', count=1, created=created).put()
        data = serializers.serialize(format, SerializeModel.all())
        db.delete(SerializeModel.all().fetch(100))
        for obj in serializers.deserialize(format, data):
            obj.save()
        self.validate_state(
            ('key.name', 'name',  'count', 'created'),
            (None,       'green', 1, created),
            ('blue_key', 'blue',  4, None),
        )

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize_db_from_string(self, data):
        """
        Reloads the database with data from a string generated by
        the serialize_db_to_string method.
        """
        data = StringIO(data)
        for obj in serializers.deserialize("json", data, using=self.connection.alias):
            obj.save()

def load_from_redis(cls, sensor_id):
        r = redis.Redis(host=settings.REDIS_HOSTNAME, port=settings.REDIS_PORT, password=settings.REDIS_PASSWORD)
        try:
            redis_response = r.get('grav_{}_full'.format(sensor_id))
            serializer = serializers.deserialize('json', redis_response)
            for obj2 in serializer:
                obj = obj2.object
                return obj
        except:
            return None


##### Tilt Hydrometer Specific Models

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize_db_from_string(self, data):
        """
        Reloads the database with data from a string generated by
        the serialize_db_to_string method.
        """
        data = StringIO(data)
        for obj in serializers.deserialize("json", data, using=self.connection.alias):
            obj.save()

def test_serialization(self):
        o1 = models.VersionModel(version=Version('0.1.1'), spec=Spec('==0.1.1,!=0.1.1-alpha'))
        o2 = models.VersionModel(version=Version('0.4.3-rc3+build3'),
            spec=Spec('<=0.1.1-rc2,!=0.1.1-rc1'))

        data = serializers.serialize('json', [o1, o2])

        obj1, obj2 = serializers.deserialize('json', data)
        self.assertEqual(o1.version, obj1.object.version)
        self.assertEqual(o1.spec, obj1.object.spec)
        self.assertEqual(o2.version, obj2.object.version)
        self.assertEqual(o2.spec, obj2.object.spec)

def test_bitmaskfield_serialization_deserialization(self):
        cases = dict(
            none=None,
            regualar=42,
        )
        for case, expected_value in cases.items():
            with self.subTest(case=case):
                model = TestModel(bitmask=expected_value)
                serialized_data = serializers.serialize("xml", [model])
                deserialized_data = list(serializers.deserialize('xml', serialized_data))
                deserialized_model = deserialized_data[0].object
                self.assertEqual(expected_value, deserialized_model.bitmask)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize_db_from_string(self, data):
        """
        Reloads the database with data from a string generated by
        the serialize_db_to_string method.
        """
        data = StringIO(data)
        for obj in serializers.deserialize("json", data, using=self.connection.alias):
            obj.save()

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize_db_from_string(self, data):
        """
        Reloads the database with data from a string generated by
        the serialize_db_to_string method.
        """
        data = StringIO(data)
        for obj in serializers.deserialize("json", data, using=self.connection.alias):
            obj.save()

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is a instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def execute_fixture(name):
    filename = os.path.join(FIXTURE_DIR, name + '.json')
    connection = connections["default"]
    with connection.constraint_checks_disabled():
        with open(filename, "r") as stream:
            objects = serializers.deserialize("json", stream)
            for obj in objects:
                if hasattr(obj.object, "_cache_data"):
                    obj.object._cache_data()
                obj.save()

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def object_version(self):
        """The stored version of the model."""
        data = self.serialized_data
        data = force_text(data.encode("utf8"))
        return list(serializers.deserialize(self.format, data, ignorenonexistent=True))[0]

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def import_(src):
    "Imports data from a Sentry export."

    from django.core import serializers
    for obj in serializers.deserialize("json", src, stream=True, use_natural_keys=True):
        obj.save()

def deserialize(format, stream_or_string, **options):
    """
    Deserialize a stream or a string. Returns an iterator that yields ``(obj,
    m2m_relation_dict)``, where ``obj`` is an instantiated -- but *unsaved* --
    object, and ``m2m_relation_dict`` is a dictionary of ``{m2m_field_name :
    list_of_related_objects}``.
    """
    d = get_deserializer(format)
    return d(stream_or_string, **options)

def deserialize_db_from_string(self, data):
        """
        Reloads the database with data from a string generated by
        the serialize_db_to_string method.
        """
        data = StringIO(data)
        for obj in serializers.deserialize("json", data, using=self.connection.alias):
            obj.save()