创建一个支持 json 序列化的类以与 Celery 一起使用

我正在使用 Celery 来运行一些后台任务。其中一项任务返回我创建的 python 类。考虑到有关使用 pickle 的警告，我想使用 json 来序列化和反序列化此类。

有没有一种简单的内置方法可以实现这一目标？

该类非常简单，它包含 3 个属性，所有属性都是命名元组的列表。它包含一些对属性执行一些计算的方法。

我的想法是序列化/反序列化 3 个属性，因为它定义了类。

这是我对编码器的想法，但我不确定如何再次解码数据？

import json

class JSONSerializable(object):
    def __repr__(self):
        return json.dumps(self.__dict__)

class MySimpleClass(JSONSerializable):
    def __init__(self, p1, p2, p3): # I only care about p1, p2, p3
        self.p1 = p1
        self.p2 = p2
        self.p3 = p2
        self.abc = p1 + p2 + p2

    def some_calc(self):
        ...

首先但并非最不重要的是：针对 pickle 的警告主要是如果您可以让第三部分在您的工作流上注入 pickle 数据。如果您确定自己的系统正在创建所有要使用的腌制数据，则根本不存在安全问题。至于兼容性，如果您的 Pickle 文件的生产者和消费者使用相同的 Python 版本，那么它相对容易处理，并且是自动的。

也就是说，对于 JSON，您必须创建一个子类蟒蛇的json.JSONEncoder and json.JSONDecoder https://docs.python.org/3/library/json.html#encoders-and-decoders- 每一个都需要作为cls论据你所有的json.dump(s) and json.load(s) calls.

一个建议是，default编码器上的方法对类进行编码__module__, its __name__和一个标识符键，比如说__custom__确保它应该被自定义解码，作为字典的键，并且对象的数据作为“数据”键。

在编码器上，您检查__custom__键，然后他们使用__new__方法，并填充其字典。就像pickle一样，在类上触发的副作用__init__不会运行。

您稍后可以增强解码器和编码器，例如，它们在类中搜索__json_encode__只能处理所需属性的方法。

示例实现：

import json

class GenericJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        try:
            return super().default(obj)
        except TypeError:
            pass
        cls = type(obj)
        result = {
            '__custom__': True,
            '__module__': cls.__module__,
            '__name__': cls.__name__,
            'data': obj.__dict__ if not hasattr(cls, '__json_encode__') else obj.__json_encode__
        }
        return result


class GenericJSONDecoder(json.JSONDecoder):
    def decode(self, str):
        result = super().decode(str)
        if not isinstance(result, dict) or not result.get('__custom__', False):
            return result
        import sys
        module = result['__module__']
        if not module in sys.modules:
            __import__(module)
        cls = getattr(sys.modules[module], result['__name__'])
        if hasattr(cls, '__json_decode__'):
            return cls.__json_decode__(result['data'])
        instance = cls.__new__(cls)
        instance.__dict__.update(result['data'])
        return instance

控制台交互测试：

In [36]: class A:
    ...:     def __init__(self, a):
    ...:         self.a = a
    ...:         

In [37]: a = A('test')

In [38]: b = json.loads(json.dumps(a, cls=GenericJSONEncoder),  cls=GenericJSONDecoder)

In [39]: b.a
Out[39]: 'test'