json 序列化和反序列化用户定义类中的对象

xriantvc  于 2023-01-03  发布在  其他
关注(0)|答案(2)|浏览(166)

假设我有这样的类层次结构:

class SerializableWidget(object):
# some code

class WidgetA(SerilizableWidget):
# some code

class WidgetB(SerilizableWidget):
# some code

我希望能够将WidgetAWidgetB(以及其他可能的小部件)的示例序列化为json形式的文本文件,然后,我希望能够反序列化这些示例,而无需事先知道它们的特定类:

some_widget = deserielize_from_file(file_path) # pseudocode, doesn't have to be exactly a method like this

some_widget需要从SerilizableWidget的精确子类中构造出来,我该怎么做呢?在层次结构的每个类中,我到底需要覆盖/实现哪些方法呢?
假设上述类的所有字段都是基元类型,那么如何覆盖__to_json____from_json__方法呢?

3wabscal

3wabscal1#

您可以使用许多方法来解决这个问题。例如,将object_hookdefault参数分别用于json.loadjson.dump
您所需要的只是将类与对象的序列化版本存储在一起,然后在加载时必须使用哪个类与哪个名称的Map。
下面的例子使用dispatcher类装饰器来存储序列化时的类名和对象,并在以后反序列化时查找它。您所需要的只是每个类上的_as_dict方法来将数据转换为dict:

import json

@dispatcher
class Parent(object):
    def __init__(self, name):
        self.name = name

    def _as_dict(self):
        return {'name': self.name}

@dispatcher
class Child1(Parent):
    def __init__(self, name, n=0):
        super().__init__(name)
        self.n = n

    def _as_dict(self):
        d = super()._as_dict()
        d['n'] = self.n
        return d

@dispatcher
class Child2(Parent):
    def __init__(self, name, k='ok'):
        super().__init__(name)
        self.k = k

    def _as_dict(self):
        d = super()._as_dict()
        d['k'] = self.k
        return d

现在进行测试。首先让我们创建一个包含3个不同类型对象的列表。

>>> obj = [Parent('foo'), Child1('bar', 15), Child2('baz', 'works')]

序列化它将在每个对象中生成具有类名的数据:

>>> s = json.dumps(obj, default=dispatcher.encoder_default)
>>> print(s)
[
  {"__class__": "Parent", "name": "foo"},
  {"__class__": "Child1", "name": "bar", "n": 15},
  {"__class__": "Child2", "name": "baz", "k": "works"}
]

然后将其加载回生成正确的对象:

obj2 = json.loads(s, object_hook=dispatcher.decoder_hook)
print(obj2)
[
  <__main__.Parent object at 0x7fb6cd561cf8>, 
  <__main__.Child1 object at 0x7fb6cd561d68>,
  <__main__.Child2 object at 0x7fb6cd561e10>
]

最后,下面是dispatcher的实现:

class _Dispatcher:
    def __init__(self, classname_key='__class__'):
        self._key = classname_key
        self._classes = {} # to keep a reference to the classes used

    def __call__(self, class_): # decorate a class
        self._classes[class_.__name__] = class_
        return class_

    def decoder_hook(self, d):
        classname = d.pop(self._key, None)
        if classname:
            return self._classes[classname](**d)
        return d

    def encoder_default(self, obj):
        d = obj._as_dict()
        d[self._key] = type(obj).__name__
        return d
dispatcher = _Dispatcher()
amrnrhlw

amrnrhlw2#

我真的很喜欢@nosklo的answer,但是我想自定义模型类型保存的属性值,所以我稍微扩展了他的代码,添加了一个子注解。
(我知道这与问题没有直接关系,但是你也可以用它来序列化到json,因为它会产生dict对象。注意你的基类必须使用@dataclass注解来正确序列化--否则你可以调整这段代码来定义__as_dict__方法,就像@nosklo的答案一样)
data.csv

model_type, prop1
sub1, testfor1
sub2, testfor2

test.py

import csv
from abc import ABC
from dataclasses import dataclass

from polymorphic import polymorphic

@polymorphic(keyname="model_type")
@dataclass
class BaseModel(ABC):
    prop1: str

@polymorphic.subtype_when_(keyval="sub1")
class SubModel1(BaseModel):
    pass

@polymorphic.subtype_when_(keyval="sub2")
class SubModel2(BaseModel):
    pass

with open('data.csv') as csvfile:
    reader = csv.DictReader(csvfile, skipinitialspace=True)
    for row_data_dict in reader:
        price_req = BaseModel.deserialize(row_data_dict)
        print(price_req, '\n\tre-serialized: ', price_req.serialize())

polymorphic.py

import dataclasses
import functools
from abc import ABC
from typing import Type

# https://stackoverflow.com/a/51976115
class _Polymorphic:
    def __init__(self, keyname='__class__'):
        self._key = keyname
        self._class_mapping = {}

    def __call__(self, abc: Type[ABC]):
        functools.update_wrapper(self, abc)
        setattr(abc, '_register_subtype', self._register_subtype)
        setattr(abc, 'serialize', lambda self_subclass: self.serialize(self_subclass))
        setattr(abc, 'deserialize', self.deserialize)
        return abc

    def _register_subtype(self, keyval, cls):
        self._class_mapping[keyval] = cls

    def serialize(self, self_subclass) -> dict:
        my_dict = dataclasses.asdict(self_subclass)
        my_dict[self._key] = next(keyval for keyval, cls in self._class_mapping.items() if cls == type(self_subclass))
        return my_dict

    def deserialize(self, data: dict):
        classname = data.pop(self._key, None)
        if classname:
            return self._class_mapping[classname](**data)
        raise ValueError(f'Invalid data: {self._key} was not found or it referred to an unrecognized class')

    @staticmethod
    def subtype_when_(*, keyval: str):
        def register_subtype_for(_cls: _Polymorphic):
            nonlocal keyval
            if not keyval:
                keyval = _cls.__name__
            _cls._register_subtype(keyval, _cls)

            @functools.wraps(_cls)
            def construct_original_subclass(*args, **kwargs):
                return _cls(*args, **kwargs)

            return construct_original_subclass

        return register_subtype_for

polymorphic = _Polymorphic

Sample console output

SubModel1(prop1='testfor1') 
    re-serialized:  {'prop1': 'testfor1', 'model_type': 'sub1'}
SubModel2(prop1='testfor2') 
    re-serialized:  {'prop1': 'testfor2', 'model_type': 'sub2'}

相关问题