Source code for jivago.serialization.serialization.namedtuple_serialization_strategy

import sys

from jivago.lang.annotations import Override
from jivago.lang.stream import Stream
from jivago.serialization.serialization_strategy import SerializationStrategy


[docs] class NamedTupleSerializationStrategy(SerializationStrategy): def __init__(self, serializer: "Serializer"): self.serializer = serializer
[docs] @Override def can_handle_serialization(self, obj) -> bool: declared_type = type(obj) if sys.version_info[0:2] == (3, 6): return hasattr(declared_type, "__bases__") \ and tuple in declared_type.__bases__ \ and hasattr(declared_type, "_field_types") return hasattr(declared_type, "__bases__") and tuple in declared_type.__bases__
[docs] @Override def serialize(self, obj) -> object: return Stream(type(obj).__annotations__)\ .map(lambda k: (k, self.serializer.serialize(obj.__getattribute__(k)))) \ .toDict()