|
1 | | -from abc import ABC |
| 1 | +from abc import ABC, ABCMeta |
2 | 2 | from logging import getLogger |
3 | 3 |
|
4 | 4 | import json |
5 | 5 | import inspect |
6 | 6 | import functools |
7 | | -from typing import Union, Iterable, List, Mapping, Dict, Any |
| 7 | +from typing import TypeVar, Union, Iterable, List, Mapping, Dict, Set, Any |
8 | 8 |
|
9 | 9 | # There are some weird (probably resolvable) errors during object cloning if this is an |
10 | 10 | # instance variable of DictSerializable. |
11 | 11 | logger = getLogger(__name__) |
12 | 12 |
|
| 13 | +DictSerializableType = TypeVar("DictSerializableType", bound="DictSerializable") |
13 | 14 |
|
14 | | -class DictSerializable(ABC): |
15 | | - """A base class for objects that can be represented as a dictionary and serialized.""" |
16 | 15 |
|
17 | | - typ = NotImplemented |
18 | | - skip = set() |
| 16 | +class DictSerializableMeta(ABCMeta): |
| 17 | + """Metaclass for tracking DictSerializable type string to class mappings.""" |
| 18 | + |
| 19 | + _class: Dict[str, type] = {} |
| 20 | + |
| 21 | + def __new__(mcs, name, bases, *args, |
| 22 | + typ: str = None, skip: Set[str] = frozenset(), |
| 23 | + **kwargs): # noqa: D102 |
| 24 | + return super().__new__(mcs, name, bases, *args, **kwargs) |
| 25 | + |
| 26 | + def __init__(cls, name, bases, *args, typ: str = None, skip: Set[str] = frozenset(), **kwargs): |
| 27 | + super().__init__(name, bases, *args, **kwargs) |
| 28 | + if typ is not None: |
| 29 | + if typ in cls._class and not issubclass(cls, cls._class.get(typ)): |
| 30 | + raise ValueError(f"{cls} attempted to take typ {typ} from {cls._class.get(typ)}, " |
| 31 | + f"which is not its ancestor.") |
| 32 | + cls.typ = typ |
| 33 | + cls._class[typ] = cls |
| 34 | + elif not hasattr(cls, "typ"): |
| 35 | + cls.typ = NotImplementedError |
| 36 | + cls.skip = {x for b in bases for x in getattr(b, 'skip', {})} | skip |
| 37 | + |
| 38 | + @property |
| 39 | + def class_mapping(cls) -> Dict[str, type]: |
| 40 | + """ |
| 41 | + Return class typ string -> class map for DictSerializable and its descendants. |
| 42 | +
|
| 43 | + Note that is actually returns a copy of the internal dict to avoid accidental breakage. |
| 44 | +
|
| 45 | + Returns |
| 46 | + ------- |
| 47 | + Dict[str, type] |
| 48 | + The mapping from typ string to class |
| 49 | +
|
| 50 | + """ |
| 51 | + return cls._class.copy() |
| 52 | + |
| 53 | + |
| 54 | +class DictSerializable(ABC, metaclass=DictSerializableMeta): |
| 55 | + """A base class for objects that can be represented as a dictionary and serialized.""" |
19 | 56 |
|
20 | 57 | @classmethod |
21 | | - def from_dict(cls, d: Mapping[str, Any]) -> "DictSerializable": |
| 58 | + def from_dict(cls, d: Mapping[str, Any]) -> DictSerializableType: |
22 | 59 | """ |
23 | 60 | Reconstitute the object from a dictionary. |
24 | 61 |
|
@@ -111,13 +148,14 @@ def build(d: Mapping[str, Any]) -> "DictSerializable": |
111 | 148 | def __repr__(self) -> str: |
112 | 149 | object_dict = self.as_dict() |
113 | 150 | # as_dict() skips over keys in `skip`, but they should be in the representation. |
114 | | - skipped_keys = {x.lstrip('_') for x in vars(self) if x in self.skip} |
| 151 | + skipped_keys = {x.lstrip('_') for x in self.skip} |
115 | 152 | for key in skipped_keys: |
116 | 153 | skipped_field = getattr(self, key, None) |
117 | 154 | object_dict[key] = self._name_repr(skipped_field) |
118 | 155 | return str(object_dict) |
119 | 156 |
|
120 | | - def _name_repr(self, entity: Union[Iterable["DictSerializable"], "DictSerializable"]) -> str: |
| 157 | + def _name_repr(self, |
| 158 | + entity: Union[Iterable[DictSerializableType], DictSerializableType]) -> str: |
121 | 159 | """ |
122 | 160 | A representation of an object or a list of objects that uses the name and type. |
123 | 161 |
|
|
0 commit comments