64

datetime.datetime など、JSON でシリアル化できないオブジェクトの形式を変更するのは簡単です。

私の要件は、デバッグ目的で、 と のような基本オブジェクトから拡張された一部のカスタム オブジェクトが json 形式でシリアル化される方法を変更することdictですlist。コード :

import datetime
import json

def json_debug_handler(obj):
    print("object received:")
    print type(obj)
    print("\n\n")
    if  isinstance(obj, datetime.datetime):
        return obj.isoformat()
    elif isinstance(obj,mDict):
        return {'orig':obj , 'attrs': vars(obj)}
    elif isinstance(obj,mList):
        return {'orig':obj, 'attrs': vars(obj)}
    else:
        return None


class mDict(dict):
    pass


class mList(list):
    pass


def test_debug_json():
    games = mList(['mario','contra','tetris'])
    games.src = 'console'
    scores = mDict({'dp':10,'pk':45})
    scores.processed = "unprocessed"
    test_json = { 'games' : games , 'scores' : scores , 'date': datetime.datetime.now() }
    print(json.dumps(test_json,default=json_debug_handler))

if __name__ == '__main__':
    test_debug_json()

デモ: http://ideone.com/hQJnLy

出力:

{"date": "2013-05-07T01:03:13.098727", "games": ["mario", "contra", "tetris"], "scores": {"pk": 45, "dp": 10}}

望ましい出力:

{"date": "2013-05-07T01:03:13.098727", "games": { "orig": ["mario", "contra", "tetris"] ,"attrs" : { "src":"console"}} , "scores": { "orig": {"pk": 45, "dp": 10},"attrs": "processed":"unprocessed }}

defaultハンドラーはシリアライズ可能なオブジェクトに対して機能しませんか? そうでない場合、 toJSON メソッドを拡張クラスに追加せずに、どうすればこれをオーバーライドできますか?

また、動作しないこのバージョンの JSON エンコーダーがあります。

class JsonDebugEncoder(json.JSONEncoder):
    def default(self,obj):
        if  isinstance(obj, datetime.datetime):
            return obj.isoformat()
        elif isinstance(obj,mDict):
            return {'orig':obj , 'attrs': vars(obj)}
        elif isinstance(obj,mList):
            return {'orig':obj, 'attrs': vars(obj)}
        else:
            return json.JSONEncoder.default(self, obj)

pickle.loads object で json.dumps を使用してハックした場合pickle,__getstate__,__setstate__,、私もそれに対してオープンです。試してみましたが、うまくいきませんでした。

4

13 に答える 13

25

JSONEncoder与えられた制限で、希望する動作を実現するには、クラスを少し掘り下げる必要があるようです。以下に、カスタムメソッドを に渡すメソッドをJSONEncoderオーバーライドするカスタムを書きました。それは世界で最もクリーンなものではありませんが、オプションを考えると最高のようで、カスタマイズを最小限に抑えます.iterencodeisinstance_make_iterencode

# customencoder.py
from json.encoder import (_make_iterencode, JSONEncoder,
                          encode_basestring_ascii, FLOAT_REPR, INFINITY,
                          c_make_encoder, encode_basestring)


class CustomObjectEncoder(JSONEncoder):

    def iterencode(self, o, _one_shot=False):
        """
        Most of the original method has been left untouched.

        _one_shot is forced to False to prevent c_make_encoder from
        being used. c_make_encoder is a funcion defined in C, so it's easier
        to avoid using it than overriding/redefining it.

        The keyword argument isinstance for _make_iterencode has been set
        to self.isinstance. This allows for a custom isinstance function
        to be defined, which can be used to defer the serialization of custom
        objects to the default method.
        """
        # Force the use of _make_iterencode instead of c_make_encoder
        _one_shot = False

        if self.check_circular:
            markers = {}
        else:
            markers = None
        if self.ensure_ascii:
            _encoder = encode_basestring_ascii
        else:
            _encoder = encode_basestring
        if self.encoding != 'utf-8':
            def _encoder(o, _orig_encoder=_encoder, _encoding=self.encoding):
                if isinstance(o, str):
                    o = o.decode(_encoding)
                return _orig_encoder(o)

        def floatstr(o, allow_nan=self.allow_nan,
                     _repr=FLOAT_REPR, _inf=INFINITY, _neginf=-INFINITY):
            if o != o:
                text = 'NaN'
            elif o == _inf:
                text = 'Infinity'
            elif o == _neginf:
                text = '-Infinity'
            else:
                return _repr(o)

            if not allow_nan:
                raise ValueError(
                    "Out of range float values are not JSON compliant: " +
                    repr(o))

            return text

        # Instead of forcing _one_shot to False, you can also just
        # remove the first part of this conditional statement and only
        # call _make_iterencode
        if (_one_shot and c_make_encoder is not None
                and self.indent is None and not self.sort_keys):
            _iterencode = c_make_encoder(
                markers, self.default, _encoder, self.indent,
                self.key_separator, self.item_separator, self.sort_keys,
                self.skipkeys, self.allow_nan)
        else:
            _iterencode = _make_iterencode(
                markers, self.default, _encoder, self.indent, floatstr,
                self.key_separator, self.item_separator, self.sort_keys,
                self.skipkeys, _one_shot, isinstance=self.isinstance)
        return _iterencode(o, 0)

をサブクラス化しCustomObjectEncoderて、カスタム オブジェクトを正しくシリアル化できるようになりました。は、CustomObjectEncoderネストされたオブジェクトの処理などのクールなこともできます。

# test.py
import json
import datetime
from customencoder import CustomObjectEncoder


class MyEncoder(CustomObjectEncoder):

    def isinstance(self, obj, cls):
        if isinstance(obj, (mList, mDict)):
            return False
        return isinstance(obj, cls)

    def default(self, obj):
        """
        Defines custom serialization.

        To avoid circular references, any object that will always fail
        self.isinstance must be converted to something that is
        deserializable here.
        """
        if isinstance(obj, datetime.datetime):
            return obj.isoformat()
        elif isinstance(obj, mDict):
            return {"orig": dict(obj), "attrs": vars(obj)}
        elif isinstance(obj, mList):
            return {"orig": list(obj), "attrs": vars(obj)}
        else:
            return None


class mList(list):
    pass


class mDict(dict):
    pass


def main():
    zelda = mList(['zelda'])
    zelda.src = "oldschool"
    games = mList(['mario', 'contra', 'tetris', zelda])
    games.src = 'console'
    scores = mDict({'dp': 10, 'pk': 45})
    scores.processed = "unprocessed"
    test_json = {'games': games, 'scores': scores,
                 'date': datetime.datetime.now()}
    print(json.dumps(test_json, cls=MyEncoder))

if __name__ == '__main__':
    main()
于 2013-07-16T18:49:46.910 に答える
5

デフォルトの関数は、ダンプされるノードがネイティブにシリアル化できない場合にのみ呼び出され、mDict クラスはそのままシリアル化されます。デフォルトがいつ呼び出され、いつ呼び出されないかを示す小さなデモを次に示します。

import json

def serializer(obj):
    print 'serializer called'
    return str(obj)

class mDict(dict):
    pass

class mSet(set):
    pass

d = mDict(dict(a=1))
print json.dumps(d, default=serializer)

s = mSet({1, 2, 3,})
print json.dumps(s, default=serializer)

そして出力:

{"a": 1}
serializer called
"mSet([1, 2, 3])"

セットはネイティブにシリアライズ可能ではありませんが、dict はシリアライズ可能であることに注意してください。

m___ クラスはシリアライズ可能であるため、ハンドラーが呼び出されることはありません。

更新 #1 -----

JSON エンコーダ コードを変更できます。これを行う方法の詳細は、使用している JSON 実装によって異なります。たとえば、simplejson の場合、encode.py の関連コードは次のとおりです。

def _iterencode(o, _current_indent_level):
    ...
        for_json = _for_json and getattr(o, 'for_json', None)
        if for_json and callable(for_json):
            ...
        elif isinstance(o, list):
            ...
        else:
            _asdict = _namedtuple_as_object and getattr(o, '_asdict', None)
            if _asdict and callable(_asdict):
                for chunk in _iterencode_dict(_asdict(),
                        _current_indent_level):
                    yield chunk
            elif (_tuple_as_array and isinstance(o, tuple)):
                ...
            elif isinstance(o, dict):
                ...
            elif _use_decimal and isinstance(o, Decimal):
                ...
            else:
                ...
                o = _default(o)
                for chunk in _iterencode(o, _current_indent_level):
                    yield chunk
                ...

つまり、エンコードされているノードが認識されたベース タイプの 1 つでない場合にのみ、default を呼び出す固定の動作があります。これは、いくつかの方法のいずれかでオーバーライドできます。

1 -- 上記で行ったように JSONEncoder をサブクラス化しますが、標準の _make_iterencode の代わりに使用する関数を指定するパラメーターを初期化子に追加します。ここで、基準を満たすクラスのデフォルトを呼び出すテストを追加します。JSON モジュールを変更していないため、これはクリーンなアプローチですが、元の _make_iterencode から多くのコードを繰り返すことになります。(このアプローチの他のバリエーションには、monkeypatching _make_iterencode またはそのサブ関数 _iterencode_dict が含まれます)。

2 -- JSON モジュール ソースを変更し、__debug__定数を使用して動作を変更します。

def _iterencode(o, _current_indent_level):
    ...
        for_json = _for_json and getattr(o, 'for_json', None)
        if for_json and callable(for_json):
            ...
        elif isinstance(o, list):
            ...
        ## added code below
        elif __debug__:
            o = _default(o)
            for chunk in _iterencode(o, _current_indent_level):
                yield chunk
        ## added code above
        else:
            ...

理想的には、JSONEncoder クラスは「すべての型にデフォルトを使用する」ことを指定するパラメーターを提供しますが、そうではありません。上記は、探していることを行う単純な 1 回限りの変更です。

于 2013-05-14T22:24:23.120 に答える
4

以下を試してください。必要な出力が生成され、比較的単純に見えます。エンコーダー クラスとの唯一の実際の違いは、decode メソッドと encode メソッドの両方をオーバーライドする必要があることです (後者は、エンコーダーが処理方法を知っている型に対して呼び出されるため)。

import json
import datetime

class JSONDebugEncoder(json.JSONEncoder):
    # transform objects known to JSONEncoder here
    def encode(self, o, *args, **kw):
        for_json = o
        if isinstance(o, mDict):
            for_json = { 'orig' : o, 'attrs' : vars(o) }
        elif isinstance(o, mList):
            for_json = { 'orig' : o, 'attrs' : vars(o) }
        return super(JSONDebugEncoder, self).encode(for_json, *args, **kw)

    # handle objects not known to JSONEncoder here
    def default(self, o, *args, **kw):
        if isinstance(o, datetime.datetime):
            return o.isoformat()
        else:
            return super(JSONDebugEncoder, self).default(o, *args, **kw)


class mDict(dict):
    pass

class mList(list):
    pass

def test_debug_json():
    games = mList(['mario','contra','tetris'])
    games.src = 'console'
    scores = mDict({'dp':10,'pk':45})
    scores.processed = "unprocessed"
    test_json = { 'games' : games , 'scores' : scores , 'date': datetime.datetime.now() }
    print(json.dumps(test_json,cls=JSONDebugEncoder))

if __name__ == '__main__':
    test_debug_json()
于 2013-07-22T13:55:33.747 に答える
2

これらをオーバーライド__instancecheck__するように定義すると、次のようになります。

def strict_check(builtin):
    '''creates a new class from the builtin whose instance check
    method can be overridden to renounce particular types'''
    class BuiltIn(type):
        def __instancecheck__(self, other):
            print 'instance', self, type(other), other
            if type(other) in strict_check.blacklist:
                return False
            return builtin.__instancecheck__(other)
    # construct a class, whose instance check method is known.
    return BuiltIn('strict_%s' % builtin.__name__, (builtin,), dict())

# for safety, define it here.
strict_check.blacklist = ()

json.encoder次に、次のようにパッチを適用してオーバーライドします_make_iterencode.func_defaults

# modify json encoder to use some new list/dict attr.
import json.encoder
# save old stuff, never know when you need it.
old_defaults = json.encoder._make_iterencode.func_defaults
old_encoder = json.encoder.c_make_encoder
encoder_defaults = list(json.encoder._make_iterencode.func_defaults)
for index, default in enumerate(encoder_defaults):
    if default in (list, dict):
        encoder_defaults[index] = strict_check(default)

# change the defaults for _make_iterencode.
json.encoder._make_iterencode.func_defaults = tuple(encoder_defaults)
# disable C extension.
json.encoder.c_make_encoder = None

...あなたの例はほぼそのまま動作します:

import datetime
import json

def json_debug_handler(obj):
    print("object received:")
    print type(obj)
    print("\n\n")
    if  isinstance(obj, datetime.datetime):
        return obj.isoformat()
    elif isinstance(obj,mDict):
        # degrade obj to more primitive dict()
        # to avoid cycles in the encoding.
        return {'orig': dict(obj) , 'attrs': vars(obj)}
    elif isinstance(obj,mList):
        # degrade obj to more primitive list()
        # to avoid cycles in the encoding.
        return {'orig': list(obj), 'attrs': vars(obj)}
    else:
        return None


class mDict(dict):
    pass


class mList(list):
    pass

# set the stuff we want to process differently.
strict_check.blacklist = (mDict, mList)

def test_debug_json():
    global test_json
    games = mList(['mario','contra','tetris'])
    games.src = 'console'
    scores = mDict({'dp':10,'pk':45})
    scores.processed = "unprocessed"
    test_json = { 'games' : games , 'scores' : scores , 'date': datetime.datetime.now() }
    print(json.dumps(test_json,default=json_debug_handler))

if __name__ == '__main__':
    test_debug_json()

私が変更する必要があったのは、循環がないことを確認することでした:

    elif isinstance(obj,mDict):
        # degrade obj to more primitive dict()
        # to avoid cycles in the encoding.
        return {'orig': dict(obj) , 'attrs': vars(obj)}
    elif isinstance(obj,mList):
        # degrade obj to more primitive list()
        # to avoid cycles in the encoding.
        return {'orig': list(obj), 'attrs': vars(obj)}

これを前のどこかに追加しますtest_debug_json

# set the stuff we want to process differently.
strict_check.blacklist = (mDict, mList)

ここに私のコンソール出力があります:

>>> test_debug_json()
instance <class '__main__.strict_list'> <type 'dict'> {'date': datetime.datetime(2013, 7, 17, 12, 4, 40, 950637), 'games': ['mario', 'contra', 'tetris'], 'scores': {'pk': 45, 'dp': 10}}
instance <class '__main__.strict_dict'> <type 'dict'> {'date': datetime.datetime(2013, 7, 17, 12, 4, 40, 950637), 'games': ['mario', 'contra', 'tetris'], 'scores': {'pk': 45, 'dp': 10}}
instance <class '__main__.strict_list'> <type 'datetime.datetime'> 2013-07-17 12:04:40.950637
instance <class '__main__.strict_dict'> <type 'datetime.datetime'> 2013-07-17 12:04:40.950637
instance <class '__main__.strict_list'> <type 'datetime.datetime'> 2013-07-17 12:04:40.950637
instance <class '__main__.strict_dict'> <type 'datetime.datetime'> 2013-07-17 12:04:40.950637
object received:
<type 'datetime.datetime'>



instance <class '__main__.strict_list'> <class '__main__.mList'> ['mario', 'contra', 'tetris']
instance <class '__main__.strict_dict'> <class '__main__.mList'> ['mario', 'contra', 'tetris']
instance <class '__main__.strict_list'> <class '__main__.mList'> ['mario', 'contra', 'tetris']
instance <class '__main__.strict_dict'> <class '__main__.mList'> ['mario', 'contra', 'tetris']
object received:
<class '__main__.mList'>



instance <class '__main__.strict_list'> <type 'dict'> {'attrs': {'src': 'console'}, 'orig': ['mario', 'contra', 'tetris']}
instance <class '__main__.strict_dict'> <type 'dict'> {'attrs': {'src': 'console'}, 'orig': ['mario', 'contra', 'tetris']}
instance <class '__main__.strict_list'> <type 'dict'> {'src': 'console'}
instance <class '__main__.strict_dict'> <type 'dict'> {'src': 'console'}
instance <class '__main__.strict_list'> <type 'list'> ['mario', 'contra', 'tetris']
instance <class '__main__.strict_list'> <class '__main__.mDict'> {'pk': 45, 'dp': 10}
instance <class '__main__.strict_dict'> <class '__main__.mDict'> {'pk': 45, 'dp': 10}
instance <class '__main__.strict_list'> <class '__main__.mDict'> {'pk': 45, 'dp': 10}
instance <class '__main__.strict_dict'> <class '__main__.mDict'> {'pk': 45, 'dp': 10}
object received:
<class '__main__.mDict'>



instance <class '__main__.strict_list'> <type 'dict'> {'attrs': {'processed': 'unprocessed'}, 'orig': {'pk': 45, 'dp': 10}}
instance <class '__main__.strict_dict'> <type 'dict'> {'attrs': {'processed': 'unprocessed'}, 'orig': {'pk': 45, 'dp': 10}}
instance <class '__main__.strict_list'> <type 'dict'> {'processed': 'unprocessed'}
instance <class '__main__.strict_dict'> <type 'dict'> {'processed': 'unprocessed'}
instance <class '__main__.strict_list'> <type 'dict'> {'pk': 45, 'dp': 10}
instance <class '__main__.strict_dict'> <type 'dict'> {'pk': 45, 'dp': 10}
{"date": "2013-07-17T12:04:40.950637", "games": {"attrs": {"src": "console"}, "orig": ["mario", "contra", "tetris"]}, "scores": {"attrs": {"processed": "unprocessed"}, "orig": {"pk": 45, "dp": 10}}}
于 2013-07-17T19:05:31.147 に答える
1

を前処理してtest_json、要件に適したものにすることはできますか? 役に立たない Encode を書くよりも、python dict を操作する方が簡単です。

import datetime
import json
class mDict(dict):
    pass

class mList(list):
    pass

def prepare(obj):
    if  isinstance(obj, datetime.datetime):
        return obj.isoformat()
    elif isinstance(obj, mDict):
        return {'orig':obj , 'attrs': vars(obj)}
    elif isinstance(obj, mList):
        return {'orig':obj, 'attrs': vars(obj)}
    else:
        return obj
def preprocessor(toJson):
    ret ={}
    for key, value in toJson.items():
        ret[key] = prepare(value)
    return ret
if __name__ == '__main__':
    def test_debug_json():
        games = mList(['mario','contra','tetris'])
        games.src = 'console'
        scores = mDict({'dp':10,'pk':45})
        scores.processed = "unprocessed"
        test_json = { 'games' : games, 'scores' : scores , 'date': datetime.datetime.now() }
        print(json.dumps(preprocessor(test_json)))
    test_debug_json()
于 2018-01-29T07:55:32.700 に答える
0

JSONEncoder.encode()をオーバーライドできるはずです。

class MyEncoder(JSONEncoder):
  def encode(self, o):
    if isinstance(o, dict):
      # directly call JSONEncoder rather than infinite-looping through self.encode()
      return JSONEncoder.encode(self, {'orig': o, 'attrs': vars(o)})
    elif isinstance(o, list):
      return JSONEncoder.encode(self, {'orig': o, 'attrs': vars(o)})
    else:
      return JSONEncoder.encode(self, o)

そして、それにパッチを当てたい場合json.dumpsは、http://docs.buildbot.net/latest/reference/json-pysrc.htmljson._default_encoderのように、のインスタンスに置き換える必要がありますMyEncoder

于 2013-07-22T03:13:04.880 に答える
0

Python 3 の簡略化 (3.9 のみでテスト済み):

from json.encoder import (_make_iterencode, JSONEncoder,
                          encode_basestring_ascii, INFINITY,
                          encode_basestring)

class CustomObjectEncoder(JSONEncoder):

    def iterencode(self, o, _one_shot=False):
        """Encode the given object and yield each string
        representation as available.

        For example::

            for chunk in JSONEncoder().iterencode(bigobject):
                mysocket.write(chunk)
                
        Change from json.encoder.JSONEncoder.iterencode is setting
        _one_shot=False and isinstance=self.isinstance
        in call to `_make_iterencode`.
        And not using `c_make_encoder`.

        """
        if self.check_circular:
            markers = {}
        else:
            markers = None
        if self.ensure_ascii:
            _encoder = encode_basestring_ascii
        else:
            _encoder = encode_basestring

        def floatstr(o, allow_nan=self.allow_nan,
                _repr=float.__repr__, _inf=INFINITY, _neginf=-INFINITY):
            # Check for specials.  Note that this type of test is processor
            # and/or platform-specific, so do tests which don't depend on the
            # internals.

            if o != o:
                text = 'NaN'
            elif o == _inf:
                text = 'Infinity'
            elif o == _neginf:
                text = '-Infinity'
            else:
                return _repr(o)

            if not allow_nan:
                raise ValueError(
                    "Out of range float values are not JSON compliant: " +
                    repr(o))

            return text

        _iterencode = _make_iterencode(
                markers, self.default, _encoder, self.indent, floatstr,
                self.key_separator, self.item_separator, self.sort_keys,
                self.skipkeys, _one_shot=False, isinstance=self.isinstance)
        return _iterencode(o, 0)

サブクラスの例:

import datetime

from rdflib.term import Literal, BNode

class RDFTermEncoder(CustomObjectEncoder):
    def isinstance(self, o, cls):
        if isinstance(o, (Literal, BNode)):
            return False
        return isinstance(o, cls)
    def default(self, o):
        if isinstance(o, Literal):
            rv = {"value": o.value}
            if o.datatype is not None:
                rv["datatype"] = o.datatype
            if o.language is not None:
                rv["lang"] = o.language
            return rv
        if isinstance(o, BNode):
            return "http://localhost/bnode/" + str(o)
        if isinstance(o, datetime.datetime):
            return o.isoformat()
        if isinstance(o, datetime.date):
            return str(o)
        # Let the base class default method raise the TypeError
        return super().default(o)

私はこれを自分の仕事にうまく使用しました

db_json = json.loads(json.dumps(db_custom, cls=RDFTermEncoder))

このスレの皆さんありがとう!

于 2021-10-15T18:06:57.170 に答える