时间戳

首页 > 解决方案 > 扩展的类 dict 子类以支持强制转换和 JSON 转储，无需额外功能

python - 扩展的类 dict 子类以支持强制转换和 JSON 转储，无需额外功能

问题描述

我需要创建一个t类 dict 类的实例，该类T支持使用 将“强制转换”为真正的 dict dict(**t)，而无需恢复为 do dict([(k, v) for k, v in t.items()])。以及支持使用标准json库转储为 JSON，无需扩展普通的 JSON 编码器（即没有为default参数提供函数）。

作为t一个正常人dict，两者都有效：

import json

def dump(data):
    print(list(data.items()))
    try:
        print('cast:', dict(**data))
    except Exception as e:
        print('ERROR:', e)
    try:
        print('json:', json.dumps(data))
    except Exception as e:
        print('ERROR:', e)

t = dict(a=1, b=2)
dump(t)

印刷：

[('a', 1), ('b', 2)]
cast: {'a': 1, 'b': 2}
json: {"a": 1, "b": 2}

但是，我想t成为该类的一个实例，例如“即时”向其项目T添加一个键，因此不可能预先插入（实际上我希望显示来自一个或多个 T 实例的合并键，这个是对那个真实的、更复杂的类的简化）。default 

class T(dict):
    def __getitem__(self, key):
        if key == 'default':
           return 'DEFAULT'
        return dict.__getitem__(self, key)

    def items(self):
        for k in dict.keys(self):
            yield k, self[k]
        yield 'default', self['default']

    def keys(self):
        for k in dict.keys(self):
            yield k 
        yield 'default'

t = T(a=1, b=2)
dump(t)

这给出了：

[('a', 1), ('b', 2), ('default', 'DEFAULT')]
cast: {'a': 1, 'b': 2}
json: {"a": 1, "b": 2, "default": "DEFAULT"}

并且演员无法正常工作，因为没有“默认”键，而且我不知道要提供哪个“魔术”功能才能使演员正常工作。

当我建立T在实现的功能上collections.abc并在子类中提供所需的抽象方法时，强制转换工作：

from collections.abc import MutableMapping

class TIter:
    def __init__(self, t):
        self.keys = list(t.d.keys()) + ['default']
        self.index = 0

    def __next__(self):
        if self.index == len(self.keys):
            raise StopIteration
        res = self.keys[self.index]
        self.index += 1
        return res

class T(MutableMapping):
    def __init__(self, **kw):
        self.d = dict(**kw)

    def __delitem__(self, key):
        if key != 'default':
            del self.d[key]

    def __len__(self):
        return len(self.d) + 1

    def __setitem__(self, key, v):
        if key != 'default':
            self.d[key] = v

    def __getitem__(self, key):
        if key == 'default':
           return 'DEFAULT'
        # return None
        return self.d[key]

    def __iter__(self):
        return TIter(self)

t = T(a=1, b=2)
dump(t)

这使：

[('a', 1), ('b', 2), ('default', 'DEFAULT')]
cast: {'a': 1, 'b': 2, 'default': 'DEFAULT'}
ERROR: Object of type 'T' is not JSON serializable

JSON 转储失败，因为该转储程序无法处理 MutableMapping子类，它使用PyDict_Check.

当我尝试创建T两者的子类时dict， MutableMapping我确实得到了与仅使用子类时相同的结果dict。

我当然可以认为这是一个错误，即json转储程序尚未更新以假设（具体子类） collections.abc.Mapping是可转储的。但即使它被承认为一个错误并在未来的 Python 版本中得到修复，我认为这样的修复不会应用于旧版本的 Python。

Q1：如何使作为T, 的子类的实现 dict正确转换？
Q2：如果 Q1 没有答案，如果我创建一个返回正确值 PyDict_Check但不执行任何实际实现的 C 级类（然后创建T它的子类以及MutableMapping（我不要认为添加这样一个不完整的 C 级别的字典会起作用，但我没有尝试过），这个傻瓜json.dumps()吗？
Q3 这是让两者像第一个例子一样工作的完全错误的方法吗？

实际代码要复杂得多，它是我的ruamel.yaml库的一部分，它必须在 Python 2.7 和 Python 3.4+ 上工作。

只要我不能解决这个问题，我就必须告诉那些曾经有功能正常的 JSON 转储器（没有额外参数）的人使用：

def json_default(obj):
    if isinstance(obj, ruamel.yaml.comments.CommentedMap):
        return obj._od
    if isinstance(obj, ruamel.yaml.comments.CommentedSeq):
        return obj._lst
    raise TypeError

print(json.dumps(d, default=json_default))

，告诉他们使用与默认（往返）加载程序不同的加载程序。例如：

yaml = YAML(typ='safe')
data = yaml.load(stream)

，在类上实现一些.to_json()方法T并让用户ruamel.yaml意识到这一点

，或者回到子类化dict并告诉人们去做

 dict([(k, v) for k, v in t.items()])

这些都不是真正友好的，并且表明不可能制作一个非平凡且与标准库很好地配合的类 dict 类。

标签： pythonjsondictionarycasting

解决方案

由于这里真正的问题是真正json.dumps的默认编码器无法将MutableMapping（或ruamel.yaml.comments.CommentedMap在您的实际示例中）视为字典，而不是像您提到的那样告诉人们将default参数设置json.dumps为您的函数，您可以使用它来设置默认值参数的值，以便人们在使用您的包时不必做任何不同的事情：json_defaultfunctools.partialjson_defaultdefaultjson.dumps

from functools import partial
json.dumps = partial(json.dumps, default=json_default)

或者如果你需要允许人们指定他们自己的default参数甚至他们自己的json.JSONEncoder子类，你可以使用一个包装器，json.dumps以便它包装参数default指定的default函数和参数default指定的自定义编码器的方法，以指定的cls为准：

import inspect

class override_json_default:
    # keep track of the default methods that have already been wrapped
    # so we don't wrap them again
    _wrapped_defaults = set()

    def __call__(self, func):
        def override_default(default_func):
            def default_wrapper(o):
                o = default_func(o)
                if isinstance(o, MutableMapping):
                    o = dict(o)
                return o
            return default_wrapper

        def override_default_method(default_func):
            def default_wrapper(self, o):
                try:
                    return default_func(self, o)
                except TypeError:
                    if isinstance(o, MutableMapping):
                        return dict(o)
                    raise
            return default_wrapper

        def wrapper(*args, **kwargs):
            bound = sig.bind(*args, **kwargs)
            bound.apply_defaults()
            default = bound.arguments.get('default')
            if default:
                bound.arguments['default'] = override_default(default)
            encoder = bound.arguments.get('cls')
            if not default and not encoder:
                bound.arguments['cls'] = encoder = json.JSONEncoder
            if encoder:
                default = getattr(encoder, 'default')
                if default not in self._wrapped_defaults:
                    default = override_default_method(default)
                    self._wrapped_defaults.add(default)
                setattr(encoder, 'default', default)
            return func(*bound.args, **bound.kwargs)

        sig = inspect.signature(func)
        return wrapper

json.dumps=override_json_default()(json.dumps)

以便以下测试代码具有自定义default函数和处理datetime对象的自定义编码器，以及没有自定义default或编码器的测试代码：

from datetime import datetime

def datetime_encoder(o):
    if isinstance(o, datetime):
        return o.isoformat()
    return o

class DateTimeEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()
        return super(DateTimeEncoder, self).default(o)

def dump(data):
    print(list(data.items()))
    try:
        print('cast:', dict(**data))
    except Exception as e:
        print('ERROR:', e)
    try:
        print('json with custom default:', json.dumps(data, default=datetime_encoder))
        print('json wtih custom encoder:', json.dumps(data, cls=DateTimeEncoder))
        del data['c']
        print('json without datetime:', json.dumps(data))
    except Exception as e:
        print('ERROR:', e)

t = T(a=1, b=2, c=datetime.now())
dump(t)

都会给出正确的输出：

[('a', 1), ('b', 2), ('c', datetime.datetime(2018, 9, 15, 23, 59, 25, 575642)), ('default', 'DEFAULT')]
cast: {'a': 1, 'b': 2, 'c': datetime.datetime(2018, 9, 15, 23, 59, 25, 575642), 'default': 'DEFAULT'}
json with custom default: {"a": 1, "b": 2, "c": "2018-09-15T23:59:25.575642", "default": "DEFAULT"}
json wtih custom encoder: {"a": 1, "b": 2, "c": "2018-09-15T23:59:25.575642", "default": "DEFAULT"}
json without datetime: {"a": 1, "b": 2, "default": "DEFAULT"}

正如评论中所指出的，上面的代码使用inspect.signature，直到 Python 3.3 才可用，即使那样，inspect.BoundArguments.apply_defaults直到 Python 3.5 才可用，并且funcsigs包，Python 3.3 的反向移植inspect.signature，也没有该apply_defaults方法。为了使代码尽可能向后兼容，您可以简单地将 Python 3.5+ 的代码复制并粘贴inspect.BoundArguments.apply_defaults到您的模块中，并根据需要将其分配为inspect.BoundArgumentsafter importing的属性funcsigs：

from collections import OrderedDict

if not hasattr(inspect, 'signature'):
    import funcsigs
    for attr in funcsigs.__all__:
        setattr(inspect, attr, getattr(funcsigs, attr))

if not hasattr(inspect.BoundArguments, 'apply_defaults'):
    def apply_defaults(self):
        arguments = self.arguments
        new_arguments = []
        for name, param in self._signature.parameters.items():
            try:
                new_arguments.append((name, arguments[name]))
            except KeyError:
                if param.default is not funcsigs._empty:
                    val = param.default
                elif param.kind is funcsigs._VAR_POSITIONAL:
                    val = ()
                elif param.kind is funcsigs._VAR_KEYWORD:
                    val = {}
                else:
                    continue
                new_arguments.append((name, val))
        self.arguments = OrderedDict(new_arguments)

    inspect.BoundArguments.apply_defaults = apply_defaults

推荐阅读