datacalssesのインスタンスを含んだ値をJSONにするときのメモ

python dataclasses json singledispatch

いつものやつです。対応していない値をjson.dumps()に与えるとTypeErrorが発生するやつです。

TypeError

例えばこういうdataclassを定義して。

from dataclasses import dataclass, asdict


@dataclass
class Point:
    x: int
    y: int

json.dumpsするとTypeError

import json
p = Point(x=1, y=2)

# TypeError: Object of type Point is not JSON serializable
print(json.dumps(p))

default

いつもどおりにdefaultを渡してあげる必要があります。ちょっとめんどくさいですが、is_dataclassという関数で調べられるのでどうにかなります。

import json


def custom_default(o):
    if is_dataclass(o):
        return asdict(o)
    raise TypeError(f"{o!r} is not JSON serializable")


p = Point(x=1, y=2)
print(json.dumps(p, default=custom_default))
# {"x": 1, "y": 2}
print(json.dumps([p, p, p], default=custom_default))
# [{"x": 1, "y": 2}, {"x": 1, "y": 2}, {"x": 1, "y": 2}]

もっと手軽に

自作のjsonモジュールを手元のライブラリあるいは手元のアプリケーションに持っておくと便利です。例えばextjsonのような名前で。 (あるいは自分自身のserialize/deserializeのためのライブラリという形にしても良いかもしれません。jsonという名前を付けずに)

extjson.py

from functools import partial
from dataclasses import is_dataclass, asdict
import json
from json import *  # noqa

# from extjson import *のときに、json moduleと同じsymbolしかexportしない
__all__ = json.__all__


def _custom_default(o):
    if is_dataclass(o):
        return asdict(o)
    raise TypeError(f"{o!r} is not JSON serializable")


dumps = partial(json.dumps, default=_custom_default)
dump = partial(json.dump, default=_custom_default)

今度はjsonとおんなじ感じに使えばdumpは大丈夫。

import extjson as json

p = Point(x=1, y=2)
print(json.dumps(p))
# {"x": 1, "y": 2}
print(json.dumps([p, p, p]))
# [{"x": 1, "y": 2}, {"x": 1, "y": 2}, {"x": 1, "y": 2}]

loadに対応したい場合には何かもう少し特殊な機構が必要です(受け取ったJSONのデータは変換後のpythonオブジェクトのtypeの情報自体は持っていないので)。

他の型にも対応したものが欲しい

例えばdatetimeなど他の型にも対応したものが欲しくなるかもしれません。その都度defaultを作ったりdefaultミドルウェアみたいな形でclosureをwrapしまくるみたいなコードを書いても良いですが。もう少し手軽にしたいですね。

２つ方法があります。１つはjson.dumpsにはplainな値のdict,listしか渡さないという方法(json.dumpのdefaultの否定)。もう一つは良い感じにdispatcherのような形で振る舞いを追加できるような構造にすること。こういうときにはfunctoolsのsingledispatchが便利です。

このsingledispatchを使ってdefault関数を作ってあげると良さそうです。

from functools import singledispatch, partial

@singledispatch
def default(o):
    raise TypeError(f"{o!r} is not JSON serializable")

dumps = partial(json.dumps, default=default)
dump = partial(json.dump, default=default)

これに対してdatetimeなどの型を登録しておきます。

import datetime

default.register(datetime.datetime, lambda o: o.isoformat())

このようにするとdatetimeに対するdefaultをサポートできます。

import extjson2 as json
from datetime import datetime

print(json.dumps(datetime.now()))
# "2018-09-29T22:17:55.296232"

それではこの仕組みをそのまま他の型にも(e.g. dataclasses, iterator)と考えてみると実は無理です。

dataclassesもsingledispatchで

dataclassesの値もsingledispatchで対応するためにはちょっと工夫が必要です。failback(失敗したときの処理)を登録できるように機能を追加してみましょう。 register()に似たようなregister_callback()という関数も持つようにしてみます。

from dataclasses import is_dataclass, asdict

# dataclassesのサポート
default.register_failback(predicate=is_dataclass, emit=asdict)

今度は大丈夫です(ちなみにjson.dumpsはiterableなオブジェクトを渡したときもエラーになるって知っていました？)。

import extjson2 as json

p = Point(x=1, y=2)
print(json.dumps(p))
# {"x": 1, "y": 2}
print(json.dumps([p, p, p]))
print(json.dumps(iter([p, p, p])))
# [{"x": 1, "y": 2}, {"x": 1, "y": 2}, {"x": 1, "y": 2}]

実装は以下の様な感じです。

extjson2.py

from functools import partial, singledispatch
from typing import Iterable
from dataclasses import is_dataclass, asdict
import json
from json import *  # noqa

__all__ = json.__all__


@singledispatch
def default(o):
    for p, fn in default.failbacks:
        if p(o):
            default.register(type(o), fn)
            return fn(o)
    raise TypeError(f"{o!r} is not JSON serializable")


default.failbacks = []  # (predicate, action)
default._seen = set()


def _register_failback(*, predicate, emit, overwrite=False):
    if overwrite or predicate not in default._seen:
        pair = (predicate, emit)
        default.failbacks.append(pair)
        default._seen.add(predicate)
        return pair


default.register_failback = _register_failback

dumps = partial(json.dumps, default=default)
dump = partial(json.dump, default=default)


########################################
## setup
########################################

def _predicate_for_iterable(o):
    return isinstance(o, Iterable)


def _emit_for_iterable(o):
    return list(o)


import datetime
default.register(datetime.datetime, lambda o: o.isoformat())
default.register_failback(predicate=_predicate_for_iterable, emit=_emit_for_iterable)
default.register_failback(predicate=is_dataclass, emit=asdict)

補足的な事柄

singledispatchを使った最後の実装は、対応するemit関数を見つけたら、対応関係を新たに追加してfailbackを無駄に呼ばないようにする処理が入っているのですが。これがclosureのなかで動的に生成される型のようなものが何度もやって来る場合にはメモリーリークの原因になるかもしれません。そのへんを上手く取り扱うならLRUみたいな仕組みを入れる必要があるかもしれません。

追記

似たような話を既に書いてました（記憶にない）。