From e8b73a7ca7e304614f6c8feca2577c5403e8d6af Mon Sep 17 00:00:00 2001 From: Michael Hudson-Doyle Date: Fri, 25 Sep 2020 11:41:50 +1200 Subject: [PATCH] include path to problem when serialization fails --- subiquity/common/serialize.py | 71 +++++++++++++++++++++-------------- 1 file changed, 42 insertions(+), 29 deletions(-) diff --git a/subiquity/common/serialize.py b/subiquity/common/serialize.py index d94ee7e9..2cef8bd0 100644 --- a/subiquity/common/serialize.py +++ b/subiquity/common/serialize.py @@ -41,90 +41,103 @@ class Serializer: self.type_serializers[datetime.datetime] = self._serialize_datetime self.type_deserializers[datetime.datetime] = self._deserialize_datetime - def _scalar(self, annotation, value, metadata): - assert type(value) is annotation, "{} is not a {}".format( - value, annotation) + def _scalar(self, annotation, value, metadata, path): + assert type(value) is annotation, "at {}, {} is not a {}".format( + path, value, annotation) return value - def _walk_Union(self, meth, args, value, metadata): + def _walk_Union(self, meth, args, value, metadata, path): NoneType = type(None) - assert NoneType in args, "can only serialize Optional" + assert NoneType in args, "at {}, can only serialize Optional" args = [a for a in args if a is not NoneType] - assert len(args) == 1, "can only serialize Optional" + assert len(args) == 1, "at {}, can only serialize Optional" if value is None: return value - return meth(args[0], value, metadata) + return meth(args[0], value, metadata, path) - def _walk_List(self, meth, args, value, metadata): - return [meth(args[0], v, metadata) for v in value] + def _walk_List(self, meth, args, value, metadata, path): + return [ + meth(args[0], v, metadata, f'{path}[{i}]') + for i, v in enumerate(value) + ] - def _serialize_datetime(self, annotation, value, metadata): - assert type(value) is annotation + def _serialize_datetime(self, annotation, value, metadata, path): + assert type(value) is annotation, "at {}, {} is not a {}".format( + path, value, annotation) if metadata is not None and 'time_fmt' in metadata: return value.strftime(metadata['time_fmt']) else: return str(value) - def _serialize_field(self, field, value): - return {field.name: self.serialize(field.type, value, field.metadata)} + def _serialize_field(self, field, value, path): + path = f'{path}.{field.name}' + return { + field.name: self.serialize(field.type, value, field.metadata, path) + } - def _serialize_attr(self, annotation, value, metadata): + def _serialize_attr(self, annotation, value, metadata, path): r = {} for field in attr.fields(annotation): - r.update(self._serialize_field(field, getattr(value, field.name))) + r.update(self._serialize_field( + field, getattr(value, field.name), path)) return r - def serialize(self, annotation, value, metadata=None): + def serialize(self, annotation, value, metadata=None, path=''): if annotation is None: assert value is None return None if annotation is inspect.Signature.empty: return value if attr.has(annotation): - return self._serialize_attr(annotation, value, metadata) + return self._serialize_attr(annotation, value, metadata, path) origin = getattr(annotation, '__origin__', None) if origin is not None: args = annotation.__args__ return self.typing_walkers[origin]( - self.serialize, args, value, metadata) + self.serialize, args, value, metadata, path) if isinstance(annotation, type) and issubclass(annotation, enum.Enum): return value.name - return self.type_serializers[annotation](annotation, value, metadata) + return self.type_serializers[annotation]( + annotation, value, metadata, path) - def _deserialize_datetime(self, annotation, value, metadata): - assert type(value) is str + def _deserialize_datetime(self, annotation, value, metadata, path): + assert type(value) is str, f'at {path}' if metadata is not None and 'time_fmt' in metadata: return datetime.datetime.strptime(value, metadata['time_fmt']) else: 1/0 - def _deserialize_field(self, field, value): + def _deserialize_field(self, field, value, path): + path = f'{path}.{field.name}' return { - field.name: self.deserialize(field.type, value, field.metadata) + field.name: self.deserialize( + field.type, value, field.metadata, path) } - def _deserialize_attr(self, annotation, value, metadata): + def _deserialize_attr(self, annotation, value, metadata, path): args = {} for field in attr.fields(annotation): - args.update(self._deserialize_field(field, value[field.name])) + args.update(self._deserialize_field( + field, value[field.name], path)) return annotation(**args) - def deserialize(self, annotation, value, metadata=None): + def deserialize(self, annotation, value, metadata=None, path=''): if annotation is None: assert value is None return None if annotation is inspect.Signature.empty: return value if attr.has(annotation): - return self._deserialize_attr(annotation, value, metadata) + return self._deserialize_attr(annotation, value, metadata, path) origin = getattr(annotation, '__origin__', None) if origin is not None: args = annotation.__args__ return self.typing_walkers[origin]( - self.deserialize, args, value, metadata) + self.deserialize, args, value, metadata, path) if isinstance(annotation, type) and issubclass(annotation, enum.Enum): return getattr(annotation, value) - return self.type_deserializers[annotation](annotation, value, metadata) + return self.type_deserializers[annotation]( + annotation, value, metadata, path) def to_json(self, annotation, value): return json.dumps(self.serialize(annotation, value))