summaryrefslogtreecommitdiffstats
path: root/src/pybind/mgr/rook/rook_client/_helper.py
blob: 382d04c27ab69601ffc5ca0e85ba35e14bb9b76b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import logging
import sys
try:
    from typing import List, Dict, Any, Optional
except ImportError:
    pass

logger = logging.getLogger(__name__)

# Tricking mypy to think `_omit`'s type is NoneType
# To make us not add things like `Union[Optional[str], OmitType]`
NoneType = type(None)
_omit = None  # type: NoneType
_omit = object()  # type: ignore


# Don't add any additionalProperties to objects. Useful for completeness testing
STRICT = False

def _str_to_class(cls, typ_str):
    if isinstance(typ_str, str):
        return getattr(sys.modules[cls.__module__], typ_str)
    return typ_str


def _property_from_json(cls, data, breadcrumb, name, py_name, typ_str, required, nullable):
    if not required and name not in data:
        return _omit
    try:
        obj = data[name]
    except KeyError as e:
        raise ValueError('KeyError in {}: {}'.format(breadcrumb, e))
    if nullable and obj is None:
        return obj
    typ = _str_to_class(cls, typ_str)
    if issubclass(typ, CrdObject) or issubclass(typ, CrdObjectList):
        return typ.from_json(obj, breadcrumb + '.' + name)
    return obj


class CrdObject(object):
    _properties = []  # type: List

    def __init__(self, **kwargs):
        for prop in self._properties:
            setattr(self, prop[1], kwargs.pop(prop[1]))
        if kwargs:
            raise TypeError(
                '{} got unexpected arguments {}'.format(self.__class__.__name__, kwargs.keys()))
        self._additionalProperties = {}  # type: Dict[str, Any]

    def _property_impl(self, name):
        obj = getattr(self, '_' + name)
        if obj is _omit:
            raise AttributeError(name + ' not found')
        return obj

    def _property_to_json(self, name, py_name, typ_str, required, nullable):
        obj = getattr(self, '_' + py_name)
        typ = _str_to_class(self.__class__, typ_str)
        if issubclass(typ, CrdObject) or issubclass(typ, CrdObjectList):
            if nullable and obj is None:
                return obj
            if not required and obj is _omit:
                return obj
            return obj.to_json()
        else:
            return obj

    def to_json(self):
        # type: () -> Dict[str, Any]
        res = {p[0]: self._property_to_json(*p) for p in self._properties}
        res.update(self._additionalProperties)
        return {k: v for k, v in res.items() if v is not _omit}

    @classmethod
    def from_json(cls, data, breadcrumb=''):
        try:
            sanitized = {
                p[1]: _property_from_json(cls, data, breadcrumb, *p) for p in cls._properties
            }
            extra = {k:v for k,v in data.items() if k not in sanitized}
            ret = cls(**sanitized)
            ret._additionalProperties = {} if STRICT else extra
            return ret
        except (TypeError, AttributeError, KeyError):
            logger.exception(breadcrumb)
            raise


class CrdClass(CrdObject):
    @classmethod
    def from_json(cls, data, breadcrumb=''):
        kind = data['kind']
        if kind != cls.__name__:
            raise ValueError("kind mismatch: {} != {}".format(kind, cls.__name__))
        return super(CrdClass, cls).from_json(data, breadcrumb)

    def to_json(self):
        ret = super(CrdClass, self).to_json()
        ret['kind'] = self.__class__.__name__
        return ret


class CrdObjectList(list):
    # Py3: Replace `Any` with `TypeVar('T_CrdObject', bound='CrdObject')`
    _items_type = None  # type: Optional[Any]

    def to_json(self):
        # type: () -> List
        if self._items_type is None:
            return self
        if issubclass(self._items_type, CrdObject) or issubclass(self._items_type, CrdObjectList):
            return [e.to_json() for e in self]
        return list(self)


    @classmethod
    def from_json(cls, data, breadcrumb=''):
        if cls._items_type is None:
            return cls(data)
        if issubclass(cls._items_type, CrdObject) or issubclass(cls._items_type, CrdObjectList):
            return cls(cls._items_type.from_json(e, breadcrumb + '[{}]'.format(i)) for i, e in enumerate(data))
        return cls(data)

    def __repr__(self):
        return '{}({})'.format(self.__class__.__name__, repr(list(self)))