namedtuple
序列化为 json 并保留字段名称的推荐方法是什么?
将
namedtuple
序列化为 json 只会导致值被序列化,而字段名称会在翻译中丢失。我希望在 json 化时也保留这些字段,因此执行了以下操作:
class foobar(namedtuple('f', 'foo, bar')):
__slots__ = ()
def __iter__(self):
yield self._asdict()
上面的内容按照我的预期序列化为 json,并且在我使用的其他地方(属性访问等)表现得像
namedtuple
,除了在迭代时出现非元组的结果(这适合我的用例)。
转换为json并保留字段名称的“正确方法”是什么?
如果它只是一个
namedtuple
,您希望序列化,则使用其 _asdict()
方法即可(使用 Python >= 2.7)
>>> from collections import namedtuple
>>> import json
>>> FB = namedtuple("FB", ("foo", "bar"))
>>> fb = FB(123, 456)
>>> json.dumps(fb._asdict())
'{"foo": 123, "bar": 456}'
这非常棘手,因为
namedtuple()
是一个返回从 tuple
派生的新类型的工厂。一种方法是让您的类也继承自 UserDict.DictMixin
,但 tuple.__getitem__
已经定义,并且需要一个表示元素位置的整数,而不是其属性的名称:
>>> f = foobar('a', 1)
>>> f[0]
'a'
从本质上讲,namedtuple 与 JSON 很奇怪,因为它实际上是一个 自定义构建的类型,其键名称作为类型定义的一部分固定,这与键名称存储在实例内部的字典不同。这可以防止您“往返”命名元组,例如如果没有其他信息,您无法将字典解码回命名元组,例如字典中特定于应用程序的类型标记
{'a': 1, '#_type': 'foobar'}
,这有点hacky。
这并不理想,但是如果您只需要将命名元组编码到字典中,另一种方法是扩展或修改您的JSON编码器以特殊情况这些类型。这是一个子类化 Python
json.JSONEncoder
的示例。这解决了确保嵌套命名元组正确转换为字典的问题:
from collections import namedtuple
from json import JSONEncoder
class MyEncoder(JSONEncoder):
def _iterencode(self, obj, markers=None):
if isinstance(obj, tuple) and hasattr(obj, '_asdict'):
gen = self._iterencode_dict(obj._asdict(), markers)
else:
gen = JSONEncoder._iterencode(self, obj, markers)
for chunk in gen:
yield chunk
class foobar(namedtuple('f', 'foo, bar')):
pass
enc = MyEncoder()
for obj in (foobar('a', 1), ('a', 1), {'outer': foobar('x', 'y')}):
print enc.encode(obj)
{"foo": "a", "bar": 1}
["a", 1]
{"outer": {"foo": "x", "bar": "y"}}
看起来您过去可以对
simplejson.JSONEncoder
进行子类化以使其工作,但使用最新的 simplejson 代码,情况不再如此:您必须实际修改项目代码。我认为 simplejson 没有理由不支持命名元组,所以我分叉了该项目,添加了命名元组支持,并且我当前正在等待我的分支被拉回主项目。如果您现在需要修复,只需从我的叉子中取出即可。
编辑:看起来最新版本的
simplejson
现在通过namedtuple_as_object
选项原生支持此功能,默认为True
。
我为此编写了一个库:https://github.com/ltwof/typedload
它可以往返命名元组。
它支持相当复杂的嵌套结构,包括列表、集合、枚举、联合、默认值。它应该涵盖大多数常见情况。
编辑:该库还支持数据类和属性类。
有一个更方便的解决方案是使用装饰器(它使用受保护的字段
_fields
)。
Python 2.7+:
import json
from collections import namedtuple, OrderedDict
def json_serializable(cls):
def as_dict(self):
yield OrderedDict(
(name, value) for name, value in zip(
self._fields,
iter(super(cls, self).__iter__())))
cls.__iter__ = as_dict
return cls
#Usage:
C = json_serializable(namedtuple('C', 'a b c'))
print json.dumps(C('abc', True, 3.14))
# or
@json_serializable
class D(namedtuple('D', 'a b c')):
pass
print json.dumps(D('abc', True, 3.14))
Python 3.6.6+:
import json
from typing import TupleName
def json_serializable(cls):
def as_dict(self):
yield {name: value for name, value in zip(
self._fields,
iter(super(cls, self).__iter__()))}
cls.__iter__ = as_dict
return cls
# Usage:
@json_serializable
class C(NamedTuple):
a: str
b: bool
c: float
print(json.dumps(C('abc', True, 3.14))
使用原生 python json 库不可能正确序列化命名元组。它将始终将元组视为列表,并且不可能覆盖默认序列化器来更改此行为。如果对象是嵌套的,情况会更糟。
最好使用更强大的库,例如 orjson:
import orjson
from typing import NamedTuple
class Rectangle(NamedTuple):
width: int
height: int
def default(obj):
if hasattr(obj, '_asdict'):
return obj._asdict()
rectangle = Rectangle(width=10, height=20)
print(orjson.dumps(rectangle, default=default))
=>
{
"width":10,
"height":20
}
它递归地将namedTuple数据转换为json。
print(m1)
## Message(id=2, agent=Agent(id=1, first_name='asd', last_name='asd', mail='[email protected]'), customer=Customer(id=1, first_name='asd', last_name='asd', mail='[email protected]', phone_number=123123), type='image', content='text', media_url='h.com', la=123123, ls=4512313)
def reqursive_to_json(obj):
_json = {}
if isinstance(obj, tuple):
datas = obj._asdict()
for data in datas:
if isinstance(datas[data], tuple):
_json[data] = (reqursive_to_json(datas[data]))
else:
print(datas[data])
_json[data] = (datas[data])
return _json
data = reqursive_to_json(m1)
print(data)
{'agent': {'first_name': 'asd',
'last_name': 'asd',
'mail': '[email protected]',
'id': 1},
'content': 'text',
'customer': {'first_name': 'asd',
'last_name': 'asd',
'mail': '[email protected]',
'phone_number': 123123,
'id': 1},
'id': 2,
'la': 123123,
'ls': 4512313,
'media_url': 'h.com',
'type': 'image'}
jsonplus 库为 NamedTuple 实例提供序列化器。如果需要,可以使用其兼容模式输出简单对象,但更喜欢默认模式,因为它有助于解码回来。
这是一个老问题了。然而:
对于所有有同样问题的人的建议是,仔细考虑使用
NamedTuple
的任何私有或内部功能,因为它们以前有过,并且会随着时间的推移再次发生变化。
例如,如果您的
NamedTuple
是一个平面值对象,并且您只对序列化它感兴趣,而不是对将其嵌套到另一个对象中的情况感兴趣,那么您可以避免删除 __dict__
所带来的麻烦或者 _as_dict()
改变并执行类似的操作(是的,这是 Python 3,因为这个答案是针对当前的):
from typing import NamedTuple
class ApiListRequest(NamedTuple):
group: str="default"
filter: str="*"
def to_dict(self):
return {
'group': self.group,
'filter': self.filter,
}
def to_json(self):
return json.dumps(self.to_dict())
我尝试使用
default
可调用 kwarg 到 dumps
以便执行 to_dict()
调用(如果可用),但没有被调用,因为 NamedTuple
可转换为列表。
这是我对这个问题的看法。它序列化 NamedTuple,处理折叠的 NamedTuple 和其中的列表
def recursive_to_dict(obj: Any) -> dict:
_dict = {}
if isinstance(obj, tuple):
node = obj._asdict()
for item in node:
if isinstance(node[item], list): # Process as a list
_dict[item] = [recursive_to_dict(x) for x in (node[item])]
elif getattr(node[item], "_asdict", False): # Process as a NamedTuple
_dict[item] = recursive_to_dict(node[item])
else: # Process as a regular element
_dict[item] = (node[item])
return _dict
simplejson.dump()
而不是 json.dump
可以完成这项工作。不过可能会慢一些。
我知道这是一个非常古老的线程,但是,我针对这个问题提出的一个解决方案是使用类似的自定义函数来修补或覆盖函数
json.encoder._make_iterencode
,其中我们将其扩展为单独处理命名元组。我不确定这是否是好的做法,或者是否有标准的更安全的方法来进行修补:
def _make_iterencode(markers, _default, _encoder, _indent, _floatstr,
_key_separator, _item_separator, _sort_keys, _skipkeys, _one_shot,
ValueError=ValueError,
dict=dict,
float=float,
id=id,
int=int,
isinstance=isinstance,
list=list,
str=str,
tuple=tuple,
_intstr=int.__repr__,
):
if _indent is not None and not isinstance(_indent, str):
_indent = ' ' * _indent
def _iterencode_list(lst, _current_indent_level):
if not lst:
yield '[]'
return
if markers is not None:
markerid = id(lst)
if markerid in markers:
raise ValueError("Circular reference detected")
markers[markerid] = lst
buf = '['
if _indent is not None:
_current_indent_level += 1
newline_indent = '\n' + _indent * _current_indent_level
separator = _item_separator + newline_indent
buf += newline_indent
else:
newline_indent = None
separator = _item_separator
first = True
for value in lst:
if first:
first = False
else:
buf = separator
if isinstance(value, str):
yield buf + _encoder(value)
elif value is None:
yield buf + 'null'
elif value is True:
yield buf + 'true'
elif value is False:
yield buf + 'false'
elif isinstance(value, int):
# Subclasses of int/float may override __repr__, but we still
# want to encode them as integers/floats in JSON. One example
# within the standard library is IntEnum.
yield buf + _intstr(value)
elif isinstance(value, float):
# see comment above for int
yield buf + _floatstr(value)
else:
yield buf
# EDIT
##################
if isinstance(value, tuple) and hasattr(value, '_asdict'):
value = value._asdict()
chunks = _iterencode_dict(value, _current_indent_level)
##################
elif isinstance(value, (list, tuple)):
chunks = _iterencode_list(value, _current_indent_level)
elif isinstance(value, dict):
chunks = _iterencode_dict(value, _current_indent_level)
else:
chunks = _iterencode(value, _current_indent_level)
yield from chunks
if newline_indent is not None:
_current_indent_level -= 1
yield '\n' + _indent * _current_indent_level
yield ']'
if markers is not None:
del markers[markerid]
def _iterencode_dict(dct, _current_indent_level):
if not dct:
yield '{}'
return
if markers is not None:
markerid = id(dct)
if markerid in markers:
raise ValueError("Circular reference detected")
markers[markerid] = dct
yield '{'
if _indent is not None:
_current_indent_level += 1
newline_indent = '\n' + _indent * _current_indent_level
item_separator = _item_separator + newline_indent
yield newline_indent
else:
newline_indent = None
item_separator = _item_separator
first = True
if _sort_keys:
items = sorted(dct.items())
else:
items = dct.items()
for key, value in items:
if isinstance(key, str):
pass
# JavaScript is weakly typed for these, so it makes sense to
# also allow them. Many encoders seem to do something like this.
elif isinstance(key, float):
# see comment for int/float in _make_iterencode
key = _floatstr(key)
elif key is True:
key = 'true'
elif key is False:
key = 'false'
elif key is None:
key = 'null'
elif isinstance(key, int):
# see comment for int/float in _make_iterencode
key = _intstr(key)
elif _skipkeys:
continue
else:
raise TypeError(f'keys must be str, int, float, bool or None, '
f'not {key.__class__.__name__}')
if first:
first = False
else:
yield item_separator
yield _encoder(key)
yield _key_separator
if isinstance(value, str):
yield _encoder(value)
elif value is None:
yield 'null'
elif value is True:
yield 'true'
elif value is False:
yield 'false'
elif isinstance(value, int):
# see comment for int/float in _make_iterencode
yield _intstr(value)
elif isinstance(value, float):
# see comment for int/float in _make_iterencode
yield _floatstr(value)
else:
# EDIT
###############
if isinstance(value, tuple) and hasattr(value, '_asdict'):
value = value._asdict()
chunks = _iterencode_dict(value, _current_indent_level)
###############
elif isinstance(value, (list, tuple)):
chunks = _iterencode_list(value, _current_indent_level)
elif isinstance(value, dict):
chunks = _iterencode_dict(value, _current_indent_level)
else:
chunks = _iterencode(value, _current_indent_level)
yield from chunks
if newline_indent is not None:
_current_indent_level -= 1
yield '\n' + _indent * _current_indent_level
yield '}'
if markers is not None:
del markers[markerid]
def _iterencode(o, _current_indent_level):
if isinstance(o, str):
yield _encoder(o)
elif o is None:
yield 'null'
elif o is True:
yield 'true'
elif o is False:
yield 'false'
elif isinstance(o, int):
# see comment for int/float in _make_iterencode
yield _intstr(o)
elif isinstance(o, float):
# see comment for int/float in _make_iterencode
yield _floatstr(o)
# EDIT
##################
elif isinstance(o, tuple) and hasattr(o, '_asdict'):
o = o._asdict()
yield from _iterencode_dict(o, _current_indent_level)
##################
elif isinstance(o, (list, tuple)):
yield from _iterencode_list(o, _current_indent_level)
elif isinstance(o, dict):
yield from _iterencode_dict(o, _current_indent_level)
else:
if markers is not None:
markerid = id(o)
if markerid in markers:
raise ValueError("Circular reference detected")
markers[markerid] = o
o = _default(o)
yield from _iterencode(o, _current_indent_level)
if markers is not None:
del markers[markerid]
return _iterencode
# alters the json lib
json.encoder._make_iterencode = _make_iterencode