Mediocre ideas in Python: serialization of attrs objects using Amazon Ion

attrsion.py

import collections
import datetime
import decimal
import importlib
import io

import amazon.ion.core
import amazon.ion.simple_types
import amazon.ion.simpleion
import attr

class SerializationError(Exception):
    pass

def serialize(obj):
    """
    Serialize an attrs object into a dict, recursively.
    This output dict is suitable for encoding with Amazon Ion.
    Supported data types are:
        * attrs-decorated classes
        * list
        * tuple
        * set
        * dict
        * str
        * bytes
        * int
        * float
        * type(None)
        * decimal.Decimal
        * datetime.datetime
    """
    if getattr(obj, '__attrs_attrs__', None):
        serialized = collections.OrderedDict()
        serialized['__attrs_class__'] = [obj.__module__, obj.__class__.__name__]
        for attribute in attr.fields(obj.__class__):
            serialized[attribute.name] = serialize(getattr(obj, attribute.name))
        return serialized
    elif isinstance(obj, (list, tuple, set)):
        return [serialize(x) for x in obj]
    elif isinstance(obj, dict):
        return collections.OrderedDict(((k, serialize(v)) for (k, v) in obj.items()))
    elif isinstance(obj, (str, int, float, decimal.Decimal, bytes, type(None), datetime.datetime)):
        return obj
    else:
        raise SerializationError('Unknown type found during serialization: {}'.format(type(obj)))

def deserialize(obj):
    """
    Deserialize an object that was serialized using `serialize`.
    Additionally, handles the null and timestamp types from amazon-ion-python.
    """
    if isinstance(obj, dict) and '__attrs_class__' in obj:
        deserialized = collections.OrderedDict()
        # TODO handle new 2-tuple class
        module_name, cls_name = obj['__attrs_class__']
        module = importlib.import_module(module_name)
        cls = getattr(module, cls_name)
        cls_fields = fields(cls) # just to raise NotAnAttrsClassError if not attrs-decorated
        for k, v in obj.items():
            if k != '__attrs_class__':
                deserialized[k] = deserialize(v)
        return cls(**deserialized)
    elif isinstance(obj, (list, tuple, set)):
        return [deserialize(x) for x in obj]
    elif isinstance(obj, dict):
        return collections.OrderedDict(((k, deserialize(v)) for (k, v) in obj.items()))
    elif amazon.ion.simple_types.is_null(obj):
        return None
    elif isinstance(obj, amazon.ion.core.Timestamp):
        return datetime.datetime(obj.year, obj.month, obj.day, obj.hour, obj.minute, obj.second, obj.microsecond, obj.tzinfo)
    elif isinstance(obj, (str, int, float, decimal.Decimal, bytes, datetime.datetime)):
        return obj
    else:
        raise SerializationError('Unknown type found during deserialization: {}'.format(type(obj)))

@attr.s
class Base:
    def serialize(self):
        return serialize(self)

    @classmethod
    def deserialize(cls, data):
        deserialized = deserialize(data)
        if cls != deserialized.__class__:
            raise SerializationError('Tried to deserialize {} into a {} object'.format(
                deserialized.__class__.__name__, cls.__name__
            ))
        return deserialized

    def dump(self, fp, binary=True):
        amazon.ion.simpleion.dump(self.serialize(), fp, binary=binary)

    @classmethod
    def load(cls, fp):
        return cls.deserialize(amazon.ion.simpleion.load(fp))

    def dumps(self, binary=True):
        buffer = io.BytesIO()
        self.dump(buffer, binary)
        if binary:
            return buffer.getvalue()
        else:
            return buffer.getvalue().decode('utf-8')

    @classmethod
    def loads(cls, data):
        if isinstance(data, str):
            buffer = io.StringIO(data)
        else:
            buffer = io.BytesIO(data)
        return cls.load(buffer)

@attr.s
class Container(Base):
    stuff = attr.ib()

@attr.s
class Datum(Base):
    value = attr.ib()



c = Container(stuff=[Datum(datetime.datetime.utcnow()), Datum(None), Datum(b'banas12!\x00')])
print('Original:', c)

print()
dumped = c.dumps()
dumped_text = c.dumps(binary=False)

print('Text form,', len(dumped_text), 'bytes:', dumped_text)
print()
print('Binary form,', len(dumped), 'bytes:', dumped)
print()


print('Reconstituted:', Container.loads(dumped_text))