serde.py

from typing import List, Dict, Union, Any
import json
import attr


_missing = dict()
_none_type = type(None)


def _item_ctor_wrap(item_ctor):
    if attr.has(item_ctor):
        return lambda data: from_json(item_ctor, data)
    else:
        return item_ctor


def _is_optional(origin, args):
    return origin is Union and len(args) == 2 and args[1] is _none_type


def from_json(cls, data):
    fields = attr.fields(cls)
    kwargs = {}
    for field in fields:
        if field.type is None:
            raise ValueError("attribute type is required")

        origin = getattr(field.type, '__origin__', None)
        args = getattr(field.type, '__args__', None)
        value = data.get(field.name, _missing)

        if field.default is not attr.NOTHING and value is _missing:  # field has a default value and is missing
            continue
        elif _is_optional(origin, args) and value is _missing:  # field is optional and is missing
            value = None
        else:
            if origin and args:
                if origin is List:  # field is a list
                    item_ctor = _item_ctor_wrap(args[0])
                    value = [item_ctor(item) for item in value]
                elif origin is Dict:  # field is a dictionary
                    key_ctor = args[0]
                    item_ctor = _item_ctor_wrap(args[1])
                    value = {key_ctor(key): item_ctor(item) for key, item in value.items()}
                elif _is_optional(origin, args):  # field is optional, but the value is present
                    item_ctor = _item_ctor_wrap(args[0])
                    value = item_ctor(value)
                else:
                    raise TypeError("unsupported type " + str(field.type))
            elif field.type is not Any:  # field is not any, call the type constructor
                item_ctor = _item_ctor_wrap(field.type)
                value = item_ctor(value)

        if value is _missing:
            value = None
        kwargs[field.name] = value
    return cls(**kwargs)


class _AttrsJSONEncoder(json.JSONEncoder):
    def default(self, o):
        if attr.has(o.__class__):
            return attr.asdict(o)
        else:
            super().default(o)


def to_json(obj, indent=None):
    if not attr.has(obj.__class__):
        raise ValueError("object must be an attrs object")
    encoder = _AttrsJSONEncoder(indent=indent)
    return encoder.encode(obj)