问题描述
我正在使用 JSON 序列化器辅助函数来轻松访问字典（基本上以 JSON 形式接收）对象。
jsondict.py
"""Utilities for working with JSON and json-like structures - deeply nested Python dicts and lists
This lets us iterate over child nodes and access elements with a dot-notation.
"""
import sys
isPy3 = sys.version_info[0]==3
if isPy3:
def __alt_str__(v,enc='utf8'):
return v if isinstance(v,bytes) else v.encode(enc)
__strTypes__ = (str,bytes)
else:
__alt_str__ = unicode
__strTypes__ = (str,unicode)
class MyLocals(object):
pass
mylocals = MyLocals()
def setErrorCollect(collect):
mylocals.error_collect = collect
setErrorCollect(False)
def errorValue(x):
if isinstance(x,__strTypes__):
return repr(x) if ' ' in x else x
return 'None' if x is None else str(x)
def condJSON(v,__name__=''):
return JSONDict(v,__name__=__name__) if isinstance(v,dict) else JSONList(v,list) else v
def condJSONSafe(v,__name__=''):
return JSONDictSafe(v,dict) else JSONListSafe(v,list) else v
class JSONListIter(object):
def __init__(self,lst,conv):
self.lst = lst
self.i = -1
self.conv = conv
def __iter__(self):
return self
def next(self):
if self.i<len(self.lst)-1:
self.i += 1
return self.conv(self.lst[self.i])
else:
raise stopiteration
if isPy3:
__next__ = next
del next
class JSONList(list):
def __init__(self,v,__name__=''):
list.__init__(self,v)
self.__name__ = __name__
def __getitem__(self,x):
return condJSON(list.__getitem__(self,x),__name__='%s\t%s'%(self.__name__,errorValue(x)))
def __iter__(self):
return JSONListIter(self,condJSON)
class JSONListSafe(JSONList):
def __getitem__(self,x):
__name__='%s\t%s'%(self.__name__,errorValue(x))
try:
return condJSONSafe(list.__getitem__(self,__name__=__name__)
except:
if mylocals.error_collect:
mylocals.error_collect(__name__)
return JSONStrSafe('')
def __iter__(self):
return JSONListIter(self,condJSONSafe)
class JSONStrSafe(str):
def __getattr__(self,attr):
return self
__getitem__ = __getattr__
class JSONDict(dict):
"Allows dotted access"
def __new__(cls,*args,**kwds):
__name__ = kwds.pop('__name__')
self = dict.__new__(cls,**kwds)
self.__name__ = __name__
return self
def __init__(self,**kwds):
kwds.pop('__name__','')
dict.__init__(self,**kwds)
def __getattr__(self,attr,default=None):
if attr in self:
return condJSON(self[attr],errorValue(attr)))
elif __alt_str__(attr) in self:
return condJSON(self[__alt_str__(attr)],errorValue(attr)))
elif attr=='__safe__':
return JSONDictSafe(self,__name__=self.__name__)
else:
raise AttributeError("No attribute or key named '%s'" % attr)
def sorted_items(self,accept=None,reject=lambda i: i[0]=='__name__'):
if accept or reject:
if not accept:
f = lambda i: not reject(i)
elif not reject:
f = accept
else: #both
f = lambda i: accept(i) and not reject(i)
return sorted(((k,condJSON(v,__name__==k)) for k,v in self.iteritems() if f((k,v))))
else:
return sorted(((k,v in self.iteritems()))
def sorted_keys(self):
return sorted(self.keys())
class JSONDictSafe(JSONDict):
"Allows dotted access"
def __getattr__(self,default=None):
if attr in self:
return condJSONSafe(self[attr],errorValue(attr)))
elif __alt_str__(attr) in self:
return condJSONSafe(self[__alt_str__(attr)],errorValue(attr)))
elif attr=='__safe__':
return self
else:
return JSONStrSafe('')
def __getitem__(self,errorValue(x))
try:
return condJSONSafe(dict.__getitem__(self,__name__=__name__)
except KeyError:
if mylocals.error_collect:
mylocals.error_collect(__name__)
return JSONStrSafe('')
def sorted_items(self,condJSONSafe(v,v in self.iteritems()))
如果 JSON 对象如下所示。
data = {'name': 'john','age': 20,'address': {'city':'xyz','country':'XZ','zip': 1223}}
json_obj = condJSONSafe(data)
我可以使用点表示法访问数据。
print(json_obj.name) --> john
print(json_obj.address.country) --> XZ
我从 JSON 中提取了一定数量的数据（在使用上述辅助函数将其作为点表示法可访问的数据之后）并将其存储到单独的列表中，例如列表 a、b、c。
然后，我传入了多处理线程，
with mp.Pool(processes=mp.cpu_count()) as pool:
res = pool.starmap(self.process_records,zip(self.a,self.b,self.c))
pool.join()
结束
TypeError: 'JSONStrSafe' object is not callable
我尝试了 this 答案，但对我不起作用。感谢您的帮助。提前致谢。
编辑： 重现示例：
test.py
import jsondict
import multiprocessing as mp
import itertools
def process_records(data,Metadata):
print(data.name)
print(Metadata)
#code to requirment
if __name__ == '__main__':
data = {
"Metadata": "test_data","cust_list": [
{
'name': 'john','address': {
'city':'xyz','zip': 1223
}
},{
'name': 'michal','age': 25,'address': {
'city':'abc','country':'CX','zip': 3435
}
},{
'name': 'david','age': 30,'address': {
'city':'mnl','country':'TD','zip': 6767
}
}
]
}
json_obj = jsondict.condJSONSafe(data)
print(json_obj.Metadata) #will print 'test_data'
print(json_obj.cust_list[0].name) #will print 'john'
print(json_obj.cust_list[2].address.city) #will print 'mnl'
with mp.Pool(processes=mp.cpu_count()) as pool:
res = pool.starmap(process_records,zip(json_obj.cust_list,itertools.repeat(json_obj.Metadata))) # --> not working
#res = pool.map(process_records,itertools.repeat(json_obj.Metadata))) --> not working
#res = [pool.apply_async(process_records,d,json_obj.Metadata) for d in json_obj.cust_list] --> not working
#apply --> not working
pool.join()
输出：
test_data
john
mnl
Traceback (most recent call last):
File "c:/Users/mohanlal/Desktop/Mock/json_err/test_app.py",line 53,in <module>
res = pool.starmap(process_records,itertools.repeat(json_obj.Metadata))) # --> not working
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\pool.py",line 268,in starmap
return self._map_async(func,iterable,starmapstar,chunksize).get()
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\pool.py",line 608,in get
raise self._value
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\pool.py",line 385,in _handle_tasks
put(task)
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\connection.py",line 206,in send
self._send_bytes(_ForkingPickler.dumps(obj))
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\reduction.py",line 51,in dumps
cls(buf,protocol).dump(obj)
TypeError: 'JSONStrSafe' object is not callable
尝试使用 startmap、map、apply_async、apply，得到相同的错误。
我已经尝试使用上面附加链接的类似问题中给出的解决方案。在出现此错误的地方修改如下。
import re
dunder_pattern = re.compile("__.*__")
protected_pattern = re.compile("_.*")
classjsONStrSafe(str):
def__getattr__(self,attr):
if dunder_pattern.match(attr) or protected_pattern.match(attr):
return super().__getattr__(attr)
return self
def__getstate__(self): returnself.__dict__
def__setstate__(self,d): self.__dict__.update(d)
__getitem__ = __getattr__
但问题仍然存在。
正如评论中所建议的，我在 getattr 的所有 3 个地方都进行了更改并尝试了。得到不同的错误如下
Process SpawnPoolWorker-1:
Traceback (most recent call last):
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\process.py",line 249,in _bootstrap
self.run()
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\process.py",line 93,in run
self._target(*self._args,**self._kwargs)
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\pool.py",line 108,in worker
task = get()
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\queues.py",line 345,in get
return _ForkingPickler.loads(res)
File "c:\Users\mohanlal\Desktop\Mock\json_err\jsondict.py",line 89,in __new__
__name__ = kwds.pop('__name__')
Process SpawnPoolWorker-2:
Process SpawnPoolWorker-4:
Traceback (most recent call last):
Traceback (most recent call last):
KeyError: '__name__'
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\process.py",in get
return _ForkingPickler.loads(res)
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\process.py",in _bootstrap
self.run()
File "c:\Users\mohanlal\Desktop\Mock\json_err\jsondict.py",in __new__
__name__ = kwds.pop('__name__')
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\process.py",**self._kwargs)
KeyError: '__name__'
File "C:\Users\mohanlal\AppData\Local\Programs\Python\python36\lib\multiprocessing\pool.py",in __new__
__name__ = kwds.pop('__name__')
KeyError: '__name__'
解决方法
问题是您处于“泡菜”中。原谅双关语——你有一个pickle 问题。当您进行多处理时，您的工作函数/方法的参数被pickled。通常，用于序列化和反序列化状态的默认值是可以的，但在您的情况下则不然。见404。序列化和反序列化对象的默认保存和加载操作是：
def save(obj):
return (obj.__class__,obj.__dict__)
def load(cls,attributes):
obj = cls.__new__(cls)
obj.__dict__.update(attributes)
return obj
请注意，在反序列化对象时，不会调用对象的 __init__ 方法，而是调用其 __new__ 方法，这就是问题所在。我不得不修改类 __new__ 的 JSONDict 方法以尝试识别它是通过反序列化调用的，因此 '__name__' 可能不存在于关键字参数中，然后不得不向该类添加自定义 __getstate__ 和 __setstate__ 方法以覆盖它保存和恢复对象属性的默认方式（方法 __init__ 保持不变）：
class JSONDict(dict):
"Allows dotted access"
def __new__(cls,*args,**kwds):
self = dict.__new__(cls,**kwds)
if kwds and '__name__' in kwds:
__name__ = kwds.pop('__name__')
self.__name__ = __name__
return self
def __init__(self,**kwds):
kwds.pop('__name__','')
dict.__init__(self,**kwds)
def __getstate__(self):
return self.__dict__
def __setstate__(self,d):
self.__dict__ = d
""" The other methods remain unmodified """
打印：
test_data
john
mnl
john
test_data
michal
david
test_data
test_data
更新
我一直在摸索为什么有必要提供 __getstate__ 和 __setstate__ pickle 方法，因为无论如何，它们所做的应该是默认操作。如果您修改程序只是为了测试酸洗，甚至没有通过插入以下行运行 Pool 方法：
json_obj = condJSONSafe(data)
# insert this line:
import pickle; print(pickle.dumps(json_obj)); sys.exit(0)
它打印：
Traceback (most recent call last):
File "test.py",line 205,in <module>
import pickle; print('pickle'); print(pickle.dumps(json_obj)); sys.exit(0)
TypeError: 'JSONStrSafe' object is not callable
在正确的位置添加打印语句后，问题很明显出在类 __getattr__ 的 JSONDictSafe 方法中。当 pickle 检查类是否实现了方法 __getstate__ 和 __setstate__ 时，当没有实现时 __getattr__ 最终被调用并作为这些属性的默认值返回一个 { {1}} 个实例。因此，与其像我那样通过定义这些方法来提供这些属性，不如添加一个简单的检查，如下所示：
JSONStrSafe