python - 扩展的类似 dict 的子类以支持转换和 JSON 转储而无需额外的

Question

我需要创建一个类字典类 T 的实例 t 支持都使用 dict(**t)“转换”到一个真正的字典，而不是恢复做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 的实例，它添加了例如一个key default "on the fly" 到它的项目，所以不可能预先插入(实际上我想要合并的键从 T 的一个或多个实例中出现，这是对真实的简化，更复杂，类)。

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"}

并且转换不能正常工作，因为没有“默认”键，而且我不知道要提供哪个“魔术”功能来进行类型转换工作。

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

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 在 C 级别明确测试。

当我试图让 T 成为 dict 和MutableMapping，我确实得到了与仅使用时相同的结果dict 子类。

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

Q1:我怎样才能使 T 实现成为dict，如何正确转换？
Q2:如果Q1没有答案，会不会如果我创建一个返回正确值的 C 级类，则可以工作PyDict_Check 但不执行任何实际实现(和然后将 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)

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

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

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

没有一个是真正友好的，并且表明这是不可能的制作一个不平凡且与标准配合良好的类似字典的类图书馆。

Answer 1

因为这里真正的问题是 json.dumps 的默认编码器无法考虑 MutableMapping(或 ruamel.yaml.comments.CommentedMap 在你的真实世界的例子中)作为一个字典，而不是告诉人们将 json.dumps 的 default 参数设置为你的 json_default 函数正如您提到的，您可以使用 functools.partial 使 json_default 成为 json.dumps default 参数的默认值code> 以便人们在使用您的包时不必做任何不同的事情:

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

或者，如果您需要允许人们指定他们自己的 default 参数，甚至是他们自己的 json.JSONEncoder 子类，您可以使用 json 的包装器。 dumps 以便它包装 default 参数指定的 default 函数和 default 指定的自定义编码器方法code>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 才可用package 是 Python 3.3 的 inspect.signature 的反向移植，也没有 apply_defaults 方法。为了使代码尽可能向后兼容，您可以简单地复制并粘贴 Python 3.5+ 的 inspect.BoundArguments.apply_defaults 的代码。到您的模块，并在根据需要导入 funcsigs 后将其分配为 inspect.BoundArguments 的属性:

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