#########################################################
# Copyright (C) 2022 SiMa Technologies, Inc.
#
# This material is SiMa proprietary and confidential.
#
# This material may not be copied or distributed without
# the express prior written permission of SiMa.
#
# All rights reserved.
#########################################################
# Code owner: Christopher Rodrigues
#########################################################
"""
An encoder/decoder interface for mapping Python objects to data structures
that are handled natively by pyyaml. See the documentation on Codec for
a more detailed description of details about the interface.
"""
from dataclasses import dataclass
from typing import List, Sequence, Tuple, Dict, Optional, TypeVar, Generic, Any, cast, Callable, Set, AbstractSet, Union
from typing_extensions import Protocol
import yaml
import numpy as np
# Types that the YAML library encodes as a YAML scalar
[docs]
SCALAR_TYPES = (int, bool, float, str, type(None))
_A = TypeVar("_A")
_B = TypeVar("_B")
_Key = TypeVar("_Key")
_Value = TypeVar("_Value")
_Scalar = TypeVar("_Scalar", *SCALAR_TYPES)
[docs]
EncState = TypeVar("EncState") # Encoder state
[docs]
DecState = TypeVar("DecState") # Decoder state
[docs]
EncStateV = TypeVar("EncStateV", contravariant=True) # Encoder state, contravariant
[docs]
DecStateV = TypeVar("DecStateV", contravariant=True) # Decoder state, contravariant
[docs]
class EncodeError(Exception):
"""
An error during YAML encoding due to invalid input or failure to handle the given input.
"""
pass
[docs]
class DecodeError(Exception):
"""
An error during YAML decoding due to invalid input or failure to handle the given input.
"""
pass
[docs]
def expect_mapping(permitted_keys: Set[str], required_keys: Set[str], doc: Any, context: str) -> None:
"""
Verify that the document is a mapping with the given keys, and raise a DecodeError otherwise.
:param permitted_keys: Keys that may be keys of the mapping
:param required_keys: Keys that must be keys of the mapping. Must be a subset of permitted_keys.
:param doc: Document to check
:param context: Description of the document, used in error messages
"""
if not isinstance(doc, dict):
raise DecodeError("Expecting a mapping in " + context)
excess = doc.keys() - permitted_keys
if excess:
raise DecodeError("Mapping has unexpected keys: " + str(excess) + " in " + context)
missing = required_keys - doc.keys()
if missing:
raise DecodeError("Mapping is missing keys: " + str(missing) + " in " + context)
[docs]
class Codec(Protocol[EncStateV, DecStateV, _A]):
"""
Protocol for an encoder/decoder that converts between a Python object and
a data structure that pyyaml can natively dump or load.
The encoded form of data must consist of data types that pyyaml handles
natively: None, Bool, int, float, str, tuple, list, and dict. Object
identity is significant, since Pyyaml uses anchors to de-duplicate
multiple references to the same object in the output. A codec may read and
write files or other persistent storage on the side.
The encode function converts a Python object to YAML. The decode function
does the reverse, and it should behave like the inverse of encode.
Data that is used by encoding or decoding is passed in the 'state' parameter.
In the intended usage, each type's codec uses its own mixin-style fields in
the state (if it needs any), so that different instances do not interact.
"""
[docs]
def encode(self, state: EncStateV, obj: _A) -> Any:
"""
Encode an object to YAML.
:param state: Encoder state used for encoding
:param obj: Object to encode
:return: Python object suitable for passing to yaml.dump
"""
raise NotImplementedError("Codec is an abstract base class")
[docs]
def decode(self, state: DecStateV, doc: Any) -> _A:
"""
Decode an object from YAML.
:param state: Encoder state used for encoding
:param doc: Python object received from yaml.load to decode
:return: Decoded object
"""
raise NotImplementedError("Codec is an abstract base class")
class _ConstCodec(Generic[EncStateV, DecStateV, _A]):
_decoded: _A
_encoded: Any
def __init__(self, decoded: _A, encoded: Any):
self._decoded = decoded
self._encoded = encoded
def encode(self, state: EncStateV, obj: _A) -> Any:
return self._encoded
def decode(self, state: DecStateV, doc: Any) -> _A:
return self._decoded
[docs]
def const_codec(decoded: _A, encoded: Any) -> Codec[EncState, DecState, _A]:
"""
An encoder that ignores its input and returns a constant value when
encoding or decoding.
:param decoded: The constant value to return when decoding.
:param encoded: The constant value to return when encoding.
:return: Constant codec for the given values
"""
return _ConstCodec(decoded, encoded)
class _MapCodec(Generic[EncStateV, DecStateV, _A, _B]):
def __init__(self, wrap: Callable[[_A], _B],
unwrap: Callable[[_B], _A],
codec: Codec[EncState, DecState, _A]):
self._wrap = wrap
self._unwrap = unwrap
self._codec = codec
def encode(self, state: EncState, obj: _B) -> Any:
return self._codec.encode(state, self._unwrap(obj))
def decode(self, state: DecState, doc: Any) -> _B:
return self._wrap(self._codec.decode(state, doc))
[docs]
def map_codec(wrap: Callable[[_A], _B],
unwrap: Callable[[_B], _A],
codec: Codec[EncState, DecState, _A]) -> Codec[EncState, DecState, _B]:
"""
Convert a codec of one type to a codec of another type by applying transformations
to the codec's input and output.
Encoding calls unwrap, then codec.encode.
Decoding calls codec.decode, then wrap.
:param wrap: Transform from the original type to the new one
:param unwrap: Transform from the new type to the original one
:param codec: Codec of the underlying type
:return: Codec of the new type
"""
return _MapCodec(wrap, unwrap, codec)
[docs]
def lazy_codec(thunk: Callable[[], Codec[EncState, DecState, _A]]) -> Codec[EncState, DecState, _A]:
"""
Lazily construct a codec.
This is intended to be used for self-referential codecs that are involved in recursive data structures, like
tree_codec = ... lazy_codec(lambda: tree_codec) ...
:param thunk: Thunk that returns a codec
:return: Codec that behaves like the thunk's return value. The thunk is evaluated the first time a method of
the codec is called.
"""
class LazyCodec:
forced_thunk: Optional[Codec[EncState, DecState, _A]]
def __init__(self):
self.forced_thunk = None
def _force(self) -> Codec[EncState, DecState, _A]:
if self.forced_thunk is None:
self.forced_thunk = thunk()
return self.forced_thunk
def encode(self, state: EncState, obj: _A) -> Any:
return self._force().encode(state, obj)
def decode(self, state: DecState, doc: Any) -> _A:
return self._force().decode(state, doc)
return LazyCodec()
class _ScalarCodec(Generic[EncStateV, DecStateV, _Scalar]):
"""
A codec for a value that pyyaml can encode as a YAML scalar.
"""
def encode(self, state: EncStateV, object: _Scalar) -> Any:
if not isinstance(object, SCALAR_TYPES):
raise EncodeError("Expected a scalar, got ", type(object))
# pyyaml can encode it directly
return object
def decode(self, state: DecStateV, doc: Any) -> _Scalar:
if not isinstance(doc, (type(None), bool, int, float, str)):
raise DecodeError("Expected a scalar")
# pyyaml can decode it directly
return cast(_Scalar, doc)
# Have to use this type synonym in scalar_codec to satisfy mypy
[docs]
ScalarCodecType = _ScalarCodec[EncStateV, DecStateV, _Scalar]
# Codec for Python values that can be encoded as YAML scalars: str, int, bool, None, and unions of these.
[docs]
scalar_codec: ScalarCodecType = _ScalarCodec()
class _EnumCodec(Generic[EncStateV, DecStateV, _A]):
def __init__(self, values: Dict[_A, str], description: str):
self._encoding = values
self._decoding = {k: s for s, k in values.items()}
# Assertion will fail if some dict keys have the same value
assert len(self._encoding) == len(self._decoding), "Encoding has duplicate values"
self._description = description
def encode(self, state: EncState, obj: _A) -> Any:
try:
return self._encoding[obj]
except KeyError:
raise EncodeError("Cannot encode {} as an enum value of {}".format(obj, self._description))
def decode(self, state: DecState, doc: Any) -> _A:
assert isinstance(doc, str)
try:
return self._decoding[doc]
except KeyError:
raise EncodeError("Cannot decode {} as an enum value of {}".format(doc, self._description))
[docs]
def enum_codec(values: Dict[_A, str], description: str) -> Codec[EncState, DecState, _A]:
"""
Create a codec for an enum, representing enum values as strings.
:param values: Mapping from enum values to strings.
:param description: A noun phrase describing this data type. It is used in error messages.
:return: Codec for enum values.
"""
return _EnumCodec(values, description)
class _ListCodec(Generic[EncStateV, DecStateV, _A]):
def __init__(self, item: Codec[EncStateV, DecStateV, _A]):
self._item = item
def encode(self, state: EncStateV, obj: List[_A]) -> Any:
# encoded representation is a list
return [self._item.encode(state, x) for x in obj]
def decode(self, state: DecStateV, doc: Any) -> List[_A]:
# encoded representation is a list
assert isinstance(doc, list)
return [self._item.decode(state, x) for x in doc]
[docs]
def list_codec(item: Codec[EncState, DecState, _A]) -> Codec[EncState, DecState, List[_A]]:
"""
Create a codec for lists.
:param item: Codec for list elements.
:return: Codec for list.
"""
return _ListCodec(item)
[docs]
def homogeneous_tuple_codec(item: Codec[EncState, DecState, _A]) -> Codec[EncState, DecState, Tuple[_A, ...]]:
"""
Create a codec for homogeneous tuples.
:param item: Codec for tuple elements.
:return: Codec for tuple.
"""
return map_codec(tuple, list, list_codec(item))
class _DictCodec(Generic[EncStateV, DecStateV, _A, _B]):
_key: Codec[EncStateV, DecStateV, _A]
_value: Codec[EncStateV, DecStateV, _B]
def __init__(self, key: Codec[EncStateV, DecStateV, _A], value: Codec[EncStateV, DecStateV, _B]):
self._key = key
self._value = value
def encode(self, state: EncStateV, obj: Dict[_A, _B]) -> Any:
key = self._key
value = self._value
# Encoded as a list of tuples to preserve order.
# The YAML library does not preserve dict order.
return [(key.encode(state, k), value.encode(state, v)) for k, v in obj.items()]
def decode(self, state: DecStateV, doc: Any) -> Dict[_A, _B]:
if not isinstance(doc, list):
raise DecodeError("Expecting a sequence to decode as a dict")
key = self._key
value = self._value
return {key.decode(state, k): value.decode(state, v) for k, v in doc}
[docs]
def dict_codec(key: Codec[EncState, DecState, _A], value: Codec[EncState, DecState, _B]) \
-> Codec[EncState, DecState, Dict[_A, _B]]:
"""
Create a codec for a dict.
:param key: Codec for dict keys
:param value: Codec for dict values
:return: Codec for dict
"""
return _DictCodec(key, value)
class _OptionalCodec(Generic[EncState, DecState, _A]):
def __init__(self, item: Codec[EncState, DecState, _A]):
self._item = item
def encode(self, state: EncState, obj: Optional[_A]):
if obj is not None:
return self._item.encode(state, obj)
return None
def decode(self, state: DecState, doc: Any) -> Optional[_A]:
if doc is not None:
return self._item.decode(state, doc)
else:
return None
[docs]
def optional_codec(item: Codec[EncState, DecState, _A]) -> Codec[EncState, DecState, Optional[_A]]:
"""
Create a codec for optional values. Since this codec assigns an interpretation to None, the
given codec should not interpret None.
:param item: Codec for values.
:return: Codec for optional values.
"""
return _OptionalCodec(item)
class _NullValueCodec(_OptionalCodec):
"""
Codec for optional values which should always have None value in the serialization.
If the decoding or encoding value is not None it will raise an exception.
"""
def __init__(self):
super().__init__(scalar_codec)
def encode(self, state: EncState, obj: Optional[_A]):
if obj is not None:
raise ValueError(f"Cannot serialize non-Null instance of {type(obj)}!")
return super().encode(state, obj)
def decode(self, state: DecState, doc: Any) -> Optional[_A]:
if doc is not None:
raise ValueError("Cannot reconstruct non-Null instance!")
return super().decode(state, doc)
[docs]
null_value_codec = _NullValueCodec()
class _MissingValueClass:
"""
Special value representing a missing dict item. This value will never occur in encoded or decoded
data.
"""
pass
_MissingValue = _MissingValueClass()
class _NamedDictCodec(Generic[EncState, DecState]):
_index: Dict[str, Codec] # Codec for each dict field
_defaults: Dict[str, Any] # Default value for each optional dict field
_required: AbstractSet[str] # Required dict fields. Equal to _index.keys() - _defaults.keys().
_description: str # Description of the data type for error messages
def __init__(self, index: Dict[str, Codec], defaults: Dict[str, Any], description: str):
assert all(k in index for k in defaults.keys()), "Default value was provided without a corresponding codec"
self._index = index
self._defaults = defaults
self._required = self._index.keys() - self._defaults.keys()
self._description = description
def encode(self, state: EncState, obj: Dict[str, Any]) -> Dict[str, Any]:
# Dict must have same keys as index
if obj.keys() != self._index.keys():
raise EncodeError("Dict does not have the expected keys in " + self._description)
result = {}
for k, codec in self._index.items():
try:
value = obj[k]
d = self._defaults.get(k, _MissingValue)
if d is not _MissingValue and d==value:
# Default value was given. Omit from the output
continue
result[k] = codec.encode(state, value)
except Exception as e:
raise EncodeError("Failed to encode field " + k + " of " + self._description) from e
return result
def decode(self, state: DecState, doc: Dict[str, Any]) -> Dict[str, Any]:
expect_mapping(self._index.keys(), self._required, doc, self._description)
result = {}
for k, codec in self._index.items():
try:
encoded = doc.get(k, _MissingValue)
if encoded is _MissingValue:
decoded = self._defaults[k]
else:
decoded = codec.decode(state, encoded)
result[k] = decoded
except Exception as e:
raise DecodeError("Failed to decode field " + k + " of " + self._description) from e
return result
[docs]
def named_dict_codec(index: Dict[str, Codec], description: str, *, defaults: Dict[str, Any] = {}) \
-> Codec[EncState, DecState, Dict[str, Any]]:
"""
Make a codec that encodes a dict having a predetermined set of string keys as a YAML mapping.
This is a typical way of encoding Python class instances.
This codec encodes and decodes Python dictionaries having the same keys as 'index'. Each dict item is
encoded and decoded using the codec for that key in 'index'. The 'defaults' dict provides default values
for the encoding. When an item compares equal to a default value, it is omitted from YAML. Missing
items in YAML are materialized using the default value.
For example, consider the following codec. ::
c = named_dict_codec({'name': scalar_codec, 'quantity': scalar_codec}, defaults = {'quantity': 1})
Calling c.encode({'name': 'x', 'quantity': 1}) returns {'name': 'x'}.
Fields are optional only in the encoded form. All fields are required in the decoded form.
:param index: A mapping from strings to codecs. Encoded dicts must have the same keys as this mapping,
and their values are encoded or decoded using these codecs.
:param description: A noun phrase describing this data type. It is used in error messages.
:param defaults: Default values for items in the index. If a default value is given, items matching that
value (when compared using __eq__) are omitted in the encoded form.
:return: Codec for dicts
"""
return _NamedDictCodec(index, defaults, description)
class _SingletonDictCodec(Generic[EncState, DecState, _A]):
_key: str
_key_set: Set[str] # A set containing only _key
_codec: Codec[EncState, DecState, _A]
_description: str # Description of the data type
def __init__(self, key: str, codec: Codec[EncState, DecState, _A], description: str):
self._key = key
self._key_set = {key}
self._codec = codec
self._description = description
def encode(self, state: EncState, obj: _A) -> Any:
return {self._key: self._codec.encode(state, obj)}
def decode(self, state: DecState, doc: Any) -> _A:
# Verify that it's a mapping containing _key
expect_mapping(self._key_set, self._key_set, doc, self._description)
return self._codec.decode(state, doc[self._key])
[docs]
def singleton_dict_codec(key: str, codec: Codec[EncState, DecState, _A], description: str) \
-> Codec[EncState, DecState, _A]:
"""
Make a codec that puts its encoded value in a YAML mapping. It creates a YAML mapping containing a
single item. This is useful when combined with other codecs.
Suppose there is a codec c, and c.encode(x) returns y. Then, singleton_dict_codec('data', c).encode(x)
returns {'data': y}. That contains the same result as the original codec, wrapped in a mapping.
:param key: Key to use in the YAML mapping
:param codec: Codec for the value
:param description: A noun phrase describing this data type. It is used in error messages.
:return: Codec that wraps the encoded value in a mapping
"""
return _SingletonDictCodec(key, codec, description)
class _IDCodec(Generic[EncState, DecState, _A]):
_make_new_id: Callable[[], int]
_get_dec_table: Callable[[DecState], Dict[int, int]]
def __init__(self, make_new_id: Callable[[], int], get_dec_table: Callable[[DecState], Dict[int, int]]):
self._make_new_id = make_new_id
self._get_dec_table = get_dec_table
def encode(self, state: EncState, obj: int) -> int:
# encoded representation is unchanged
return obj
def decode(self, state: DecState, doc: int) -> int:
# decoded representation is translated to a new ID
table = self._get_dec_table(state)
try:
return table[doc]
except KeyError:
# This ID has not appeared yet during decoding.
# Create a new ID, and use the created ID as the translation.
i = self._make_new_id()
table[doc] = i
return i
[docs]
def unique_id_codec(make_new_id: Callable[[], int],
get_dec_table: Callable[[DecState], Dict[int, int]]) \
-> Codec[EncState, DecState, int]:
"""
Make a codec that encodes unique integer IDs. IDs are encoded as YAML integers.
IDs are converted to new values when they are decoded so that the result is unique after decoding.
:param make_new_id: How to make a new unique ID. This is called during decoding for every ID
that appears in the YAML document.
:param get_dec_table: Get decoder state for this codec to use and modify. The state should be
initialized by calling unique_id_dec_table.
:return: codec for unique integer IDs.
"""
return _IDCodec(make_new_id, get_dec_table)
[docs]
def unique_id_dec_table() -> Dict[int, int]:
"""Create a decoder table suitable for use by unique_id_codec."""
return {}
[docs]
class AliasCodec(Generic[EncState, DecState, _A]):
_get_enc_table: Callable[[EncState], Dict[int, Any]]
_get_dec_table: Callable[[DecState], Dict[int, Any]]
_codec: Codec[EncState, DecState, _A]
def __init__(self, get_enc_table: Callable[[EncState], Dict[int, Any]],
get_dec_table: Callable[[DecState], Dict[int, Any]],
codec: Codec[EncState, DecState, _A]):
self._get_enc_table = get_enc_table
self._get_dec_table = get_dec_table
self._codec = codec
[docs]
def encode(self, state: EncState, obj: _A) -> Any:
table = self._get_enc_table(state)
identity = id(obj)
# Look up the previously encoded form of the object
memoized = table.get(identity, _MissingValue)
if memoized is not _MissingValue:
# Found previously encoded value
return memoized
# Else, object has not been encoded before. Encode and memoize it.
table[identity] = encoded = self._codec.encode(state, obj)
return encoded
[docs]
def decode(self, state: DecState, doc: Any) -> _A:
table = self._get_dec_table(state)
identity = id(doc)
# Look up the previously decoded form of this object
memoized = table.get(identity, _MissingValue)
if memoized is not _MissingValue:
# Found previously decoded value
return memoized
# Else, document has not been decoded before. Decode and memoize it.
table[identity] = decoded = self._codec.decode(state, doc)
return decoded
[docs]
def aliasable_codec(get_enc_table: Callable[[EncState], Dict[int, Any]],
get_dec_table: Callable[[DecState], Dict[int, Any]],
codec: Codec[EncState, DecState, _A]) \
-> Codec[EncState, DecState, _A]:
"""
Make a codec that applies YAML anchors to encoded data and uses anchors for repeated references to the
same object. An object will be encoded at most once and will be recreated once during decoding,
no matter how many times it appears.
This is done by memoizing the encoded representation so that the yaml library will use anchors. Thus, it
depends on the yaml library to use anchors and memoization when the yaml library encounters
the same object in its input.
:param get_enc_table: Get encoder state for this codec to use and modify. The state should be
initialized by calling aliasable_codec_enc_table.
:param get_dec_table: Get decoder state for this codec to use and modify. The state should be
initialized by calling aliasable_codec_dec_table.
:param codec: Codec to extend with aliasing
:return: Codec that uses anchors for aliasing
"""
return AliasCodec(get_enc_table, get_dec_table, codec)
[docs]
def aliasable_codec_enc_table() -> Dict[int, Any]:
"""Create an encoder table suitable for use by aliasable_codec."""
return {}
[docs]
def aliasable_codec_dec_table() -> Dict[int, Any]:
"""Create a decoder table suitable for use by aliasable_codec."""
return {}
class _TaggedUnionCodec(Generic[EncState, DecState, _A]):
_tag_name: str
_codecs: Dict[str, Codec[EncState, DecState, _A]]
_get_tag: Callable[[_A], str]
def __init__(self, tag_name: str, codecs: Dict[str, Codec[EncState, DecState, _A]],
get_tag: Callable[[_A], str]):
self._tag_name = tag_name
self._codecs = codecs
self._get_tag = get_tag
def encode(self, state: EncState, obj: _A) -> Any:
tag = self._get_tag(obj)
codec = self._codecs[tag]
encoded = codec.encode(state, obj)
# Don't overwrite an existing label
if self._tag_name in encoded:
raise EncodeError("Conflicting use of mapping key {}".format(self._tag_name))
return {self._tag_name: tag, **encoded}
def decode(self, state: DecState, doc: Any) -> _A:
if not isinstance(doc, dict):
raise DecodeError("Expected a mapping")
try:
tag = doc[self._tag_name]
except KeyError:
raise DecodeError("Missing mapping key {}".format(self._tag_name))
if not isinstance(tag, str):
raise DecodeError("Tag is not a string")
try:
codec = self._codecs[tag]
except KeyError:
raise DecodeError("Unrecognized tag {}".format(tag))
# Remove tag from the mapping. Don't modify the original mapping.
doc = doc.copy()
del doc[self._tag_name]
return codec.decode(state, doc)
[docs]
def intrusive_tagged_union_codec(tag_name: str,
codecs: Dict[str, Codec[EncState, DecState, _A]],
get_tag: Callable[[_A], str]) \
-> Codec[EncState, DecState, _A]:
"""
Make a codec for a union-like type to be encoded as a tagged union.
The encoding is a YAML mapping, where one key in the mapping has a string tag and the
other keys are tag-dependent. For example, consider the following codec for Union[int, str]. ::
value_codec = singleton_dict_codec('number', scalar_codec)
info_codec = singleton_dict_codec('message', scalar_codec)
def get_tag(x):
return 'info' if isinstance(x, str) else 'value'
c = intrusive_tagged_union_codec('type', {'value': value_codec, 'info': message_codec}, get_tag)
Calling c.encode(3) returns {'type': 'value', 'number': 3}.
Calling c.encode('Three') returns {'type': 'info', 'message': 'Three'}.
The dict key 'type' is the tag indicating which union member it is. The other keys hold the rest of the data.
:param tag_name: Name of the tag in encoded YAML mappings.
:param codecs: Alternative codecs, indexed by tag.
:param get_tag: Selects which codec to use for encoding a value.
:return: Codec for the tagged union
"""
return _TaggedUnionCodec(tag_name, codecs, get_tag)
[docs]
def type_tagged_codec(tag_name: str, codecs: Sequence[Tuple[str, type, Codec[EncState, DecState, _A]]]) \
-> Codec[EncState, DecState, _A]:
"""
Make a tagged union codec where each tag stands for a distinct type.
:param codecs: Tag, type, and codec of each alternative encoding.
:return: Codec for the tagged union
"""
type_dict = {ty: tag for tag, ty, _ in codecs}
codec_dict = {tag: codec for tag, _, codec in codecs}
def get_tag(x: _A) -> str:
t = type(x)
try:
return type_dict[t]
except KeyError:
raise EncodeError("Cannot encode data with type {}".format(t))
return intrusive_tagged_union_codec(tag_name, codec_dict, get_tag)
# Special value representing the absence of a default value in FieldEncoding
class _NoDefaultClass:
pass
_NoDefault = _NoDefaultClass()
@dataclass(frozen=True)
[docs]
class FieldEncoding:
"""
Describes the encoding of one class field as YAML data when encoding a class as a mapping.
:param python_name: Name of the field in Python source code.
:param yaml_name: Name of the field in a YAML mapping.
:param codec: Encoding of the field's value.
:param default: Field's default value, used if the field is omitted in the YAML mapping.
"""
[docs]
codec: Codec[EncState, DecState, Any]
[docs]
default: Any = _NoDefault
# Abbreviation of FieldEncoding, for convenience when making a list of fields
[docs]
def dataclass_style_codec(cls: type, fields: Sequence[FieldEncoding]) -> Codec[EncState, DecState, _A]:
"""
Make a codec that translates a class to a mapping with named fields, for classes
that resemble dataclasses. To use this codec, a class must have a dataclass-style
constructor so that it's possible to clone with this pattern of field accesses
and constructor call: ::
copy_of_instance = MyClass(instance.field1, instance.field2, instance.field3)
Python reflection is used to access fields by name.
:param cls: The Python class to create the codec for
:param fields: The class's fields
:return: Codec for the class
"""
yaml_names = [f.yaml_name for f in fields]
class_name = cls.__name__
def wrap(mapping: Dict[str, Any]) -> _A:
return cast(_A, cls(*(mapping[n] for n in yaml_names)))
def unwrap(obj: _A) -> Dict[str, Any]:
return {f.yaml_name: getattr(obj, f.python_name) for f in fields}
mapping_codec = named_dict_codec({f.yaml_name: f.codec for f in fields}, class_name,
defaults={f.yaml_name: f.default for f in fields if f.default is not _NoDefault})
return map_codec(wrap, unwrap, mapping_codec)
@dataclass(frozen=True)
[docs]
class ConstructorEncoding:
"""
An encoding of one class instance as YAML data as part of an encoding scheme for
algebraic data types.
* If a sequence of one or more field encodings is given, a dataclass_style_codec encoding is used.
* If an empty sequence is given, a const_codec encoding is used.
* If a codec is given, the codec is used to encode an instance of the class. Its encoded form must
be a mapping. The algebraic data type tag will be injected into the mapping.
:param cls: The Python class that this constructor represents
:param yaml_name: The string used to represent that class in YAML
:param fields: Encoding of the class's fields. Either a list giving the encoding of each field,
or a Codec giving the encoding of an entire class instance as a YAML mapping.
"""
[docs]
fields: Union[Sequence[FieldEncoding], Codec[EncState, DecState, Any]]
def _constructor_codec(encoding: ConstructorEncoding) -> Codec[EncState, DecState, Any]:
"""
Get a constructor's codec from ConstructorEncoding.
"""
try:
iterator = iter(encoding.fields)
except TypeError:
# Not a sequence. It must be a codec.
return encoding.fields
if not encoding.fields:
# Empty sequence. Use const codec.
return const_codec(encoding.cls(), {})
# Nonempty sequence. Use dataclass codec.
return dataclass_style_codec(encoding.cls, encoding.fields)
[docs]
def adt_style_codec(tag_name: str, members: Sequence[ConstructorEncoding]) -> Codec[EncState, DecState, _A]:
"""
Make a codec for data in the style of an algebraic data type. The decoded data is
a union of the types given in members. The encoded data is a mapping constructed with
intrusive_tagged_union_codec. It has a tag indicating which type is encoded there.
:param tag_name: The name of the type tag in the encoded dict.
:param members: The members of the data type. An instance of the type is an instance of one member.
:return: Codec for the type.
"""
return type_tagged_codec(tag_name, [(m.yaml_name, m.cls, _constructor_codec(m)) for m in members])