创建一个支持与celery一起使用的json序列化的类 [英] Create a class that support json serialization for use with Celery
问题描述
我正在使用Celery运行一些后台任务.其中一项任务返回了我创建的python类.鉴于有关使用pickle的警告，我想使用json对此类进行序列化和反序列化.
I'm using Celery to run some background tasks. One of the tasks returns a python class I created. I want to use json to serialize and deserialize this class, given the warnings about using pickle.
是否有一种简单的内置方法来实现这一目标?
Is there a simple built in way to achieve this?
该类非常简单，它包含3个属性，所有属性都是命名元组的列表.它包含一些对属性进行一些计算的方法.
The class is very simple, it contains 3 attributes all of which are lists of named tuples. It contains a couple of methods that performs some calculations on the attributes.
我的想法是对3个属性进行序列化/反序列化，因为这定义了类.
My idea is to serialize/deserialize the 3 attributes, since that defines the class.
这是我对编码器的想法，但是我不确定如何再次解码数据?
This is my idea for the encoder, but I'm not sure how to decode the data again?
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):
...
推荐答案
首先但并非最不重要的一点:关于腌渍的警告主要是如果您可以让第3部分在您的工作流中注入腌渍的数据.如果您确定自己的系统正在创建要使用的所有腌制数据，则完全没有安全问题.至于兼容性，如果您为Pickle文件的生产者和使用者使用相同的Python版本，则它相对易于处理，并且是自动的.
First but not least important: the warnings against pickle are mainly if you could have 3rd partis injecting pickled data on your worker stream. If you are certain your own system is creating all pickled data to be consumed, there is no security problem at all. And as for compatibility, it is relatively easy to handle, and automatic if you are on the same Python version for produers and consumers of your Pickle files.
也就是说，对于JSON，您必须创建 Python的json.JSONEncoder和json.JSONDecoder -每个都需要作为cls参数传递给所有json.dump(s)和json.load(s)调用.
That said, for JSON, you have to create a subclass of Python's json.JSONEncoder and json.JSONDecoder - each of which will need to be passed as the cls argument to all your json.dump(s) and json.load(s) calls.
建议编码器上的default方法对类__module__，其__name__和一个标识符键(例如__custom__)进行编码，以确保应对其进行自定义解码，作为字典的键，以及对象的数据作为数据"键.
A suggestion is that the default method on the encoder encodes the class __module__, its __name__ and a identifier key, say __custom__ to ensure it should be custom decoded, as keys to a dictionary, and the object's data as a "data" key.
然后在编码器上，检查__custom__键，然后它们使用__new__方法实例化一个类，并填充其字典.像泡菜一样，在类__init__上触发的副作用也不会运行.
And on the encoder, you check for the __custom__ key, and them instantiate a class using the __new__ method, and populate its dict. Like for pickle, side-effects that are triggered on the class __init__ won't run.
稍后，您可以增强解码器和编码器，以便例如在类中搜索只能处理所需属性的__json_encode__方法.
You can later on enhance your decoder and encoder so that, for example, they search the class for a __json_encode__ method that could handle only the desired attributes.
示例实现:
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
在控制台上进行交互式测试:
Interactive test on the console:
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'
这篇关于创建一个支持与celery一起使用的json序列化的类的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
