【问题标题】:What would be the pythonic way to go to prevent circular loop while writing JSON?在编写 JSON 时防止循环循环的 pythonic 方法是什么?
【发布时间】:2019-07-19 06:16:49
【问题描述】:

我有两个类 A 和 B,每个类都在列表中存储对另一个类的对象的引用

class A:
    def __init__(self,name):
        self.name = name
        self.my_Bs = []
    def registerB(self,b):
        self.my_Bs.append(b)

class B:
    def __init__(self,name):
        self.name = name
        self.my_As = []
    def registerA(self,a):
        self.my_As.append(a)

现在,我的应用构建了两个列表,一个是 A 的对象,一个是 B 的对象,具有交叉引用。

# a list of As, a list of Bs
list_of_As = [A('firstA'), A('secondA')]
list_of_Bs = [B('firstB'), B('secondB')]
# example of one cross-reference
list_of_As[0].registerB(list_of_Bs[1])
list_of_Bs[1].registerA(list_of_As[0])

显然,如果我在 list_of_... 上调用 json.dumps(),我会收到循环引用错误。

为了规避这个问题,我想要做的是使用 元素列表 name 属性 转储 JSON,而不是 对象本身的列表

# This is what I want to obtain for
# the JSON for list_of_As
[
    {'name' : 'firstA', 'my_Bs': ['secondB']},
    {'name' : 'secondA', 'my_Bs': []}
]

我能想到的唯一方法是在每个类中维护一个额外的字符串列表(分别为my_Bs_namesmy_As_names)并使用JSONEncoder,如下所示:

class MyEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, 'A'):
            return { # filter out the list of B objects
                k: v for k, v in obj.__dict__.items() if k != 'my_Bs'
            }
        if isinstance(obj, 'B'):
            return { # filter out the list of A objects
                k: v for k, v in obj.__dict__.items() if k != 'my_As'
            }
        return super(MyEncoder, self).default(obj)

# Use the custom encoder to dump JSON for list_of_As
print json.dumps(list_of_As, cls=MyEncoder)

如果我没记错的话,我会得到以下结果:

# This is what I obtain for
# the JSON for list_of_As with the code above
[
    {'name' : 'firstA', 'my_Bs_names': ['secondB']},
    {'name' : 'secondA', 'my_Bs_names': []}
]

有没有更优雅的方法来获得这个结果?例如,不需要任何额外的字符串列表?

【问题讨论】:

    标签: python json serialization


    【解决方案1】:

    防止循环引用错误的通用JSONEncoder

    以下编码器类MyEncoder 执行嵌套对象的递归编码,直到检测到循环引用,返回其“名称”属性而不是对象本身。

    import json
    class MyEncoder(json.JSONEncoder):
        def __init__(self, *args, **argv):
            super().__init__(*args, **argv)
            self.proc_objs = []
        def default(self, obj):
            if isinstance(obj,(A,B)):
                if obj in self.proc_objs:
                    return obj.name # short circle the object dumping
                self.proc_objs.append(obj)
                return obj.__dict__
            return obj
    
    json.dumps(list_of_As, cls=MyEncoder, check_circular=False, indent=2)
    

    输出:

    [
      { "name": "firstA",
        "my_Bs": [
          { "name": "secondB",
            "my_As": [ "firstA" ]
          }
        ]
      },
      { "name": "secondA", "my_Bs": [] }
    ]
    

    使用自定义toJSON 方法

    你可以在你的类中实现serializer method

    class JSONable:
        def toJSON(self):
            d = dict()
            for k,v in self.__dict__.items():
                # save a list of "name"s of the objects in "my_As" or "my_Bs"
                d[k] = [o.name for o in v] if isinstance(v, list) else v
            return d
    
    class A(JSONable):
        def __init__(self,name):
            self.name = name
            self.my_Bs = []
        def register(self,b):
            self.my_Bs.append(b)
    
    class B(JSONable):
        def __init__(self,name):
            self.name = name
            self.my_As = []
        def register(self,a):
            self.my_As.append(a)
    
    json.dumps(list_of_As, default=lambda x: x.toJSON(), indent=2)
    

    输出:

    [
      { "name":  "firstA",  "my_Bs": [  "secondB" ] },
      { "name":  "secondA", "my_Bs": [] }
    ]
    

    【讨论】:

    • 我不确定将您的类称为“通用”是否适用,因为您的解决方案(如我的)明确测试特定类,并且需要 .name 属性。请注意,使用列表来跟踪已经看到的对象意味着您的解决方案将随着要序列化的对象数量的每次增加而二次降级(因为列表上的in 需要 O(N) 时间并且您测试对象 N 次)。
    【解决方案2】:

    最佳实践方法是在编码时记录已经看到的对象的id() 值。 id() 值对于具有重叠生命周期的对象是唯一的,并且在编码时,您通常可以指望对象不是短暂的。这适用于任何对象类型,并且不需要对象是可散列的。

    copypickle 模块都在 memo 字典中使用此技术,该字典将 id() 值映射到它们的对象以供以后参考。

    你也可以在这里使用这种技术;你真的只需要保留一个 set 的 id 来检测你可以返回 .name 属性。使用集合可以快速有效地测试重复引用(成员资格测试需要 O(1) 恒定时间,而列表需要 O(N) 线性时间):

    class CircularEncoder(json.JSONEncoder):
        def __init__(self, *args, **kwargs):
            kwargs['check_circular'] = False  # no need to check anymore
            super(CircularEncoder, self).__init__(*args, **kwargs)
            self._memo = set()
    
        def default(self, obj):
            if isinstance(obj, (A, B)):
                d = id(obj)
                if d in self._memo:
                    return obj.name
                self._memo.add(d)
                return vars(obj)
            return super(CircularEncoder, self).default(obj)
    

    然后在这个类中使用json.dumps()

    json.dumps(list_of_As, cls=CircularEncoder)
    

    对于您的示例输入,这会产生:

    >>> print(json.dumps(list_of_As, cls=CircularEncoder, indent=2))
    [
      {
        "name": "firstA",
        "my_Bs": [
          {
            "name": "secondB",
            "my_As": [
              "firstA"
            ]
          }
        ]
      },
      {
        "name": "secondA",
        "my_Bs": []
      }
    ]
    

    【讨论】:

      【解决方案3】:

      这个怎么样?

      • AB 这样的类只需要指定一个类属性(_deep_fields)列出它们可能导致循环依赖的属性(需要“浅”-序列化)
      • 他们还需要从ShallowSerializable 继承,如果shallowTrue,它会简单地忽略_deep_fields 中的属性
      • 编码器对对象的所有键进行编码,但在所有值上调用make_shallow,以确保将shallow=True 发送到从ShallowSerializable 继承的任何对象
      • 解决方案是通用的,因为任何其他需要实现此行为的类只需要从ShallowSerializable 继承并定义_deep_fields
      class ShallowSerializable(object):
           _deep_fields = set()
           def get_dict(self,  shallow=False):
               return {
                   k: v
                   for k, v in self.__dict__.items()
                   if not shallow or k not in self._deep_fields
               }
      
      class A(ShallowSerializable):
          _deep_fields = {'my_Bs'}
      
          def __init__(self,name):
              self.name = name
              self.my_Bs = []
      
           def registerB(self,b):
              self.my_Bs.append(b)
      
      class B(ShallowSerializable):
          _deep_fields = {'my_As'}
      
          def __init__(self,name):
              self.name = name
              self.my_As = []
      
          def registerA(self,a):
              self.my_As.append(a)
      
      
      class MyEncoder(json.JSONEncoder):
          def make_shallow(self, obj):
              if isinstance(obj, ShallowSerializable):
                  return obj.get_dict(shallow=True)
              elif isinstance(obj, dict):
                  return {k: self.make_shallow(v) for k, v in obj.items()}
              elif isinstance(obj, list):
                  return [self.make_shallow(x) for x in obj]
              else:
                  return obj
      
          def default(self, obj):
              return {
                  k: self.make_shallow(v)
                  for k, v in obj.__dict__.items()
              }
      
      

      用法:

      list_of_As = [A('firstA'), A('secondA')]
      list_of_Bs = [B('firstB'), B('secondB')]
      # example of one cross-reference
      list_of_As[0].registerB(list_of_Bs[1])
      list_of_Bs[1].registerA(list_of_As[0])
      
      json.dumps(list_of_As, cls=MyEncoder)
      >>> '[{"my_Bs": [{"name": "secondB"}], "name": "firstA"}, {"my_Bs": [], "name": "secondA"}]'
      
      json.dumps(list_of_Bs, cls=MyEncoder)
      >>> '[{"my_As": [], "name": "firstB"}, {"my_As": [{"name": "firstA"}], "name": "secondB"}]'
      

      【讨论】:

        【解决方案4】:

        您可以通过更改对象的字符串表示或说通过 python 魔术方法制作的 python 对象的表示来做到这一点,这有多少库更改了它们的控制台和字符串表示,以使用类的十六进制作为回报

        Run Code Here

        import json
        class A:
            def __init__(self,name):
                self.name = name
                self.my_Bs = []
        
            def registerB(self,b):
                self.my_Bs.append(b)
        
            def __str__(self):
                _storage = {
                    "name" : self.name,
                    "my_Bs": [obj.name for obj in self.my_Bs]
                }
                return json.dumps(_storage)
        
            __repr__ = __str__
        
        class B:
            def __init__(self,name):
                self.name = name
                self.my_As = []
        
            def registerA(self,a):
                self.my_As.append(a)
        
            def __str__(self):
                _storage = {
                    "name" : self.name,
                    "my_Bs" : [obj.name for obj in self.my_As]
                }
                return json.dumps(_storage)
        
            __repr__ = __str__
        
        
        # a list of As, a list of Bs
        list_of_As = [A('firstA'), A('secondA')]
        list_of_Bs = [B('firstB'), B('secondB')]
        # example of one cross-reference
        list_of_As[0].registerB(list_of_Bs[1])
        list_of_Bs[1].registerA(list_of_As[0])
        str(list_of_As) # will make it done without  more overhead
        

        您现在还可以优化您的代码,因为它只是改变了您的表示方式,而无需额外的类包

        【讨论】:

          猜你喜欢
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 2010-12-26
          • 1970-01-01
          • 2022-06-17
          • 1970-01-01
          • 2018-04-28
          • 2016-10-10
          相关资源
          最近更新 更多