edq.util.serial

  1import collections
  2import enum
  3import os
  4import types
  5import typing
  6
  7import edq.core.errors
  8
  9import edq.util.common
 10import edq.util.enum
 11import edq.util.json
 12
 13PODType = typing.Union[bool, float, int, str, typing.List['PODType'], typing.Dict[str, 'PODType'], None]  # pylint: disable=invalid-name
 14""" A "Plain Old Data" type that can be easily represented (e.g., in JSON). """
 15
 16SerializationBaseClass = typing.TypeVar('SerializationBaseClass', bound = 'SerializationBase')
 17
 18DictSerializerClass = typing.TypeVar('DictSerializerClass', bound = 'DictSerializer')
 19DictDeserializerClass = typing.TypeVar('DictDeserializerClass', bound = 'DictDeserializer')
 20DictConverterClass = typing.TypeVar('DictConverterClass', bound = 'DictConverter')
 21
 22PODSerializerClass = typing.TypeVar('PODSerializerClass', bound = 'PODSerializer')
 23PODDeserializerClass = typing.TypeVar('PODDeserializerClass', bound = 'PODDeserializer')
 24PODConverterClass = typing.TypeVar('PODConverterClass', bound = 'PODConverter')
 25
 26# Alias the context (which we don't store here because of a cyclic dependency with edq.util.json).
 27SerializationContext = edq.util.common.SerializationContext
 28
 29# For Python 3.9 compatibility, we need to dynamically check for which union types are available.
 30_UNION_TYPES: typing.Set[typing.Any] = {typing.Union}
 31if (hasattr(types, 'UnionType')):
 32    _UNION_TYPES.add(getattr(types, 'UnionType'))
 33
 34class SerializationBase:
 35    """
 36    A base class for the serialization classes.
 37
 38    Note that this causes diamond inheritance for classes that extend *Converter,
 39    but since this class only provides simple functionality not overwritten by any branch child,
 40    there should be no issues.
 41
 42    General (but inefficient) implementations of several core Python equality, comparison, and representation methods are provided.
 43    A default hash implementation is provided, but it is up to child classes themselves to ensure they are immutable
 44    if they want to be used as set elements or dict keys.
 45    """
 46
 47    serialization_omit_empty: bool = False
 48    """
 49    Do not include empty fields in serialization.
 50    An empty field meets one of the following conditions:
 51     - Has a `__len__` method which returns 0.
 52     - Has a `_serialization_is_empty` method that returns true.
 53    """
 54
 55    serialization_omit_none: bool = False
 56    """ Do not include None (null) fields in serialization. """
 57
 58    serialization_skip_fields: typing.Union[typing.Set[str], None] = None
 59    """ A list of field names to skip. """
 60
 61    serialization_include_init_context: bool = False
 62    """ Do not send the serialization context to an object's init. """
 63
 64    serialization_error_class: typing.Type[Exception] = ValueError
 65    """ The class to use when raising errors. """
 66
 67    @classmethod
 68    def skip_field(cls, name: str, value: typing.Any) -> bool:
 69        """ Check if a field should be skipped. """
 70
 71        if ((cls.serialization_skip_fields is not None) and (name in cls.serialization_skip_fields)):
 72            return True
 73
 74        if (cls.serialization_omit_none and (value is None)):
 75            return True
 76
 77        if (cls.serialization_omit_empty and hasattr(value, '__len__') and (len(value) == 0)):
 78            return True
 79
 80        if (cls.serialization_omit_empty and hasattr(value, '_serialization_is_empty') and value._serialization_is_empty()):
 81            return True
 82
 83        return False
 84
 85    def __lt__(self, other: 'SerializationBase') -> bool:
 86        return repr(self) < repr(other)
 87
 88    def __hash__(self) -> int:
 89        return hash(repr(self))
 90
 91    def __str__(self) -> str:
 92        return repr(self)
 93
 94    def __repr__(self) -> str:
 95        return edq.util.json.dumps(self)
 96
 97    def _eq(self,
 98            other: object,
 99            self_serializer: typing.Callable,
100            other_serializer: typing.Callable,
101            context: typing.Union[SerializationContext, None] = None,
102            ) -> bool:
103        """
104        Check for equality using the given func..
105        """
106
107        if (not isinstance(other, type(self))):
108            return False
109
110        return bool(self_serializer(context) == other_serializer(context))
111
112    @classmethod
113    def _from_path(cls: typing.Type[SerializationBaseClass],
114            path: str,
115            deserializer: typing.Callable,
116            context: typing.Union[SerializationContext, None] = None,
117            ) -> SerializationBaseClass:
118        """
119        The internal helper for from_path().
120        """
121
122        path = os.path.abspath(path)
123
124        if (context is None):
125            context = SerializationContext()
126        else:
127            context = context.copy()
128
129        context.base_dir = os.path.dirname(path)
130        context.source_path = path
131
132        data = edq.util.json.load_path(path, **context.json_options)
133
134        return deserializer(data, context)  # type: ignore[no-any-return]
135
136    def _to_path(self,
137            path: str,
138            serializer: typing.Callable,
139            context: typing.Union[SerializationContext, None] = None,
140            ) -> None:
141        """ Write this object to the given path as JSON. """
142
143        if (context is None):
144            context = SerializationContext()
145        else:
146            context = context.copy()
147
148        if ((not os.path.isabs(path)) and (context.base_dir is not None)):
149            path = os.path.join(context.base_dir, path)
150
151        context.source_path = os.path.abspath(path)
152        context.base_dir = os.path.dirname(context.source_path)
153
154        data = serializer(context)
155        edq.util.json.dump_path(data, context.source_path, **context.json_options)
156
157class PODSerializer(SerializationBase):
158    """
159    A class that can represent itself as a POD type.
160    Sibling to PODDeserializer.
161    """
162
163    def to_pod(self,
164            context: typing.Union[SerializationContext, None] = None,
165            ) -> PODType:
166        """
167        Get a POD representation of this object.
168
169        The default implementation will convert to a dict (similar to a DictSerializer).
170        """
171
172        if (context is None):
173            context = SerializationContext()
174
175        data: typing.Dict[str, typing.Any] = {}
176
177        for (key, value) in vars(self).items():
178            if (self.skip_field(key, value)):
179                continue
180
181            data[key] = generic_to_pod(value, context, self.serialization_error_class)
182
183        return data
184
185    def to_path(self,
186            path: str,
187            context: typing.Union[SerializationContext, None] = None,
188            ) -> None:
189        """ Write this object to the given path as JSON. """
190
191        self._to_path(path, self.to_pod, context)
192
193    def __eq__(self, other: object) -> bool:
194        """
195        Check for equality.
196
197        This check uses to_pod() and compares the results.
198        This may not be complete or efficient depending on the child class.
199        """
200
201        return self._eq(other, self.to_pod, getattr(other, 'to_pod', None))  # type: ignore[arg-type]
202
203class PODDeserializer(SerializationBase):
204    """
205    A class that can construct itself from a POD type.
206    Sibling to PODSerializer.
207    """
208
209    @classmethod
210    def prep_init_data(cls,
211            data: typing.Dict[str, typing.Any],
212            context: typing.Union[SerializationContext, None] = None,
213            ) -> typing.Dict[str, typing.Any]:
214        """
215        Prepare data to be passed into this class' constructor.
216
217        By default, this is called by from_pod().
218        A child can override this or prep_init_data() depending on the functionality they want.
219
220        A general (but inefficient) implementation is provided by default.
221        This implementation will attempt to use type hints (of the classes constructor) to convert enums and DictDeserializers.
222        """
223
224        if (context is None):
225            context = SerializationContext()
226
227        new_data = {}
228        constructor_types = typing.get_type_hints(cls.__init__)
229
230        for (key, value) in data.items():
231            if (cls.skip_field(key, value)):
232                continue
233
234            new_data[key] = _from_pod(f"field {key}", constructor_types.get(key, None), value, context, cls.serialization_error_class)
235
236        return new_data
237
238    @classmethod
239    def from_pod(cls: typing.Type[PODDeserializerClass],
240            data: PODType,
241            context: typing.Union[SerializationContext, None] = None,
242            ) -> PODDeserializerClass:
243        """
244        Create an instance of this class from a POD.
245
246        The default implementation will call the class' constructor with one of two things:
247        a splat/unpacking (**) of the incoming data if the data is a dict,
248        otherwise the data itself.
249        """
250
251        if (context is None):
252            context = SerializationContext()
253
254        if (isinstance(data, dict)):
255            new_data = cls.prep_init_data(data, context)
256
257            if (cls.serialization_include_init_context):
258                new_data['context'] = context
259
260            return cls(**new_data)
261
262        return cls(data)  # type: ignore[call-arg]
263
264    @classmethod
265    def from_path(cls: typing.Type[PODDeserializerClass],
266            path: str,
267            context: typing.Union[SerializationContext, None] = None,
268            ) -> PODDeserializerClass:
269        """
270        Read the path (as JSON) and call from_pod().
271
272        If a serialization context is passed in to this function,
273        a copy will be made with the new base dir and source path.
274        """
275
276        return cls._from_path(path, cls.from_pod, context)
277
278class PODConverter(PODSerializer, PODDeserializer):
279    """ A PODSerializer and PODDeserializer. """
280
281    def copy(self,
282            context: typing.Union[SerializationContext, None] = None,
283            ) -> 'PODConverter':
284        """
285        Make a deep copy of this object.
286        The default implementation will use to_pod() and from_pod() to make a copy.
287
288        Callers should be cautious of fileds that are skipped in serialization,
289        e.g., via `SerializationBase.serialization_skip_fields`.
290        """
291
292        if (context is None):
293            context = SerializationContext()
294
295        return self.from_pod(self.to_pod(context), context)
296
297class DictSerializer(PODSerializer):
298    """
299    A base class for class that can represent itself as a dict.
300    The intention is that the dict can then be cleanly converted to/from JSON.
301    """
302
303    def to_dict(self,
304            context: typing.Union[SerializationContext, None] = None,
305            ) -> typing.Dict[str, PODType]:
306        """
307        Return a dict that can be used to represent this object.
308        If the dict is passed to from_dict(), an identical object should be reconstructed.
309
310        A general (but inefficient) implementation is provided by default.
311        """
312
313        data = self.to_pod(context)
314        if (not isinstance(data, dict)):
315            raise self.serialization_error_class(f"DictSerializer's to_pod() did not return a dict, found '{type(data)}'.")
316
317        return data
318
319    def to_path(self,
320            path: str,
321            context: typing.Union[SerializationContext, None] = None,
322            ) -> None:
323        """ Write this object to the given path as JSON. """
324
325        self._to_path(path, self.to_dict, context)
326
327    def __eq__(self, other: object) -> bool:
328        """
329        Check for equality.
330
331        This check uses to_pod() and compares the results.
332        This may not be complete or efficient depending on the child class.
333        """
334
335        return self._eq(other, self.to_dict, getattr(other, 'to_dict', None))  # type: ignore[arg-type]
336
337class DictDeserializer(PODDeserializer):
338    """
339    A base class for class that can reconstruct (deserialize) themselves from a dict.
340    The intention is that the class can be cleanly converted from JSON.
341
342    Any class that uses the default implementations should have a constructor
343    that accepts arguments with the same name as their members.
344    It is generally recommended to also have the constructor accept a kwargs,
345    since values will be blindly passed to the constructor.
346    """
347
348    @classmethod
349    def from_dict(cls: typing.Type[DictDeserializerClass],
350            data: typing.Dict[str, PODType],
351            context: typing.Union[SerializationContext, None] = None,
352            ) -> DictDeserializerClass:
353        """
354        Return an instance of this subclass created using the given dict.
355        If the dict came from to_dict(), the returned object should be equivalent to the original.
356
357        By default, this function just calls the class' constructor with the output of prep_init_data().
358        A child can override this or prep_init_data() depending on the functionality they want.
359
360        A general (but inefficient) implementation is provided by default.
361        This implementation will attempt to use type hints (of the classes constructor) to convert enums and DictDeserializers.
362        """
363
364        return cls.from_pod(data, context)
365
366    @classmethod
367    def from_path(cls: typing.Type[DictDeserializerClass],
368            path: str,
369            context: typing.Union[SerializationContext, None] = None,
370            ) -> DictDeserializerClass:
371        """
372        Read the path (as JSON) and call from_dict().
373
374        If a serialization context is passed in to this function,
375        a copy will be made with the new base dir and source path.
376        """
377
378        return cls._from_path(path, cls.from_dict, context)
379
380class DictConverter(PODConverter, DictSerializer, DictDeserializer):
381    """ A DictSerializer and DictDeserializer. """
382
383def _check_issubclass(allowed_type: typing.Any, target: typing.Type) -> bool:
384    """
385    Call issubclass(), but squash and type errors.
386    issubclass() can be picky about the input types is accepts.
387    """
388
389    if (allowed_type is None):
390        return False
391
392    try:
393        return issubclass(allowed_type, target)
394    except TypeError:
395        return False
396
397def generic_to_pod(
398        raw_value: typing.Any,
399        context: SerializationContext,
400        serialization_error_class: typing.Type[Exception] = ValueError,
401        ) -> PODType:
402    """
403    Attempt to convert a value to a POD type.
404
405    Has special handling for:
406        - DictSerializer
407        - PODSerializer
408        - enum.Enum
409        - (list, tuple, set)
410        - dict
411    """
412
413    if (raw_value is None):
414        return None
415
416    # Simple types that are already POD.
417    if (isinstance(raw_value, (bool, float, int, str))):
418        return raw_value
419
420    if (isinstance(raw_value, DictSerializer)):
421        return raw_value.to_dict(context)
422
423    if (isinstance(raw_value, PODSerializer)):
424        return raw_value.to_pod(context)
425
426    if (isinstance(raw_value, enum.Enum)):
427        return generic_to_pod(raw_value.value, context, serialization_error_class)
428
429    if (isinstance(raw_value, (list, tuple, set))):
430        items = [generic_to_pod(item, context, serialization_error_class) for item in raw_value]
431
432        # Sort sets for consistency.
433        if (isinstance(raw_value, set)):
434            # Sort using the string representation of the items (since the set could be heterogeneous and have non-comparable items).
435            sort_list = sorted([(item, i) for (i, item) in enumerate(map(str, items))])
436            items = [items[i] for (_, i) in sort_list]
437
438        return items
439
440    if (isinstance(raw_value, dict)):
441        return {key: generic_to_pod(value, context, serialization_error_class) for (key, value) in raw_value.items()}
442
443    raise serialization_error_class(f"Unable to convert value to simple (edq.util.serial.POD) type: '{raw_value}' (type: '{type(raw_value)}').")
444
445def _from_pod(
446        label: str,
447        type_hint: typing.Any,
448        raw_value: typing.Any,
449        context: SerializationContext,
450        serialization_error_class: typing.Type[Exception] = ValueError,
451        ) -> typing.Any:
452    """ Attempt to convert a value to the hinted value. """
453
454    try:
455        return _from_pod_internal(label, type_hint, raw_value, context, serialization_error_class)
456    except Exception as ex:
457        raise serialization_error_class(f"Failed to deserialize {label}.") from ex
458
459def _from_pod_internal(
460        label: str,
461        type_hint: typing.Any,
462        raw_value: typing.Any,
463        context: SerializationContext,
464        serialization_error_class: typing.Type[Exception] = ValueError,
465        ) -> typing.Any:
466    """ Attempt to convert a value to the hinted value. """
467
468    # If there is no type hint or anything is allowed, then just return the raw value.
469    if ((type_hint is None) or (type_hint is typing.Any)):
470        return raw_value
471
472    allowed_types: typing.Tuple[typing.Any, ...] = tuple([type_hint])
473
474    # Check if the hint is a union.
475    if (typing.get_origin(type_hint) in _UNION_TYPES):
476        allowed_types = typing.get_args(type_hint)
477
478    if (len(allowed_types) == 0):
479        return raw_value
480
481    # Check for a None early.
482    if ((raw_value is None) and (type(None) in allowed_types)):
483        return None
484
485    # Check each possible type and match the first one.
486    for allowed_type in allowed_types:
487        if (_check_issubclass(allowed_type, DictDeserializer) and isinstance(raw_value, dict)):
488            return allowed_type.from_dict(raw_value, context)
489
490        if (_check_issubclass(allowed_type, PODDeserializer)):
491            return allowed_type.from_pod(raw_value, context)
492
493        if (_check_issubclass(allowed_type, enum.Enum)):
494            if (edq.util.enum.has_value(allowed_type, raw_value)):
495                return allowed_type(raw_value)
496
497        # Sequence container types.
498        if ((typing.get_origin(allowed_type) in (list, tuple, set, collections.abc.Sequence)) and isinstance(raw_value, (list, tuple, set))):
499            collection_type = typing.get_origin(allowed_type)
500            if (collection_type is collections.abc.Sequence):
501                collection_type = list
502
503            item_type = None
504            args = typing.get_args(allowed_type)
505            if (len(args) > 0):
506                item_type = args[0]
507
508            return collection_type([
509                _from_pod(f"{label}[{i}]", item_type, item, context, serialization_error_class)
510                for (i, item)
511                in enumerate(raw_value)
512            ])
513
514        # Dict
515        if ((typing.get_origin(allowed_type) in (dict, collections.abc.Mapping)) and isinstance(raw_value, dict)):
516            key_type = None
517            value_type = None
518
519            args = typing.get_args(allowed_type)
520            if (len(args) == 2):
521                key_type = args[0]
522                value_type = args[1]
523
524            return {
525                _from_pod(f"{label}.({key} (key))", key_type, key, context, serialization_error_class):
526                _from_pod(f"{label}.{key}", value_type, value, context, serialization_error_class)
527                for (key, value)
528                in raw_value.items()
529            }
530
531    # No special conversion was made, try to force the first type.
532
533    force_type = allowed_types[0]
534    if (not isinstance(force_type, type)):
535        force_type = typing.get_origin(force_type)
536
537    if (force_type is not None):
538        return force_type(raw_value)
539
540    return raw_value