diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-04-07 18:45:59 +0000 |
commit | 19fcec84d8d7d21e796c7624e521b60d28ee21ed (patch) | |
tree | 42d26aa27d1e3f7c0b8bd3fd14e7d7082f5008dc /src/pybind/mgr/rook/rook_client/_helper.py | |
parent | Initial commit. (diff) | |
download | ceph-upstream/16.2.11+ds.tar.xz ceph-upstream/16.2.11+ds.zip |
Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r-- | src/pybind/mgr/rook/rook_client/_helper.py | 108 |
1 files changed, 108 insertions, 0 deletions
diff --git a/src/pybind/mgr/rook/rook_client/_helper.py b/src/pybind/mgr/rook/rook_client/_helper.py new file mode 100644 index 000000000..b4c7793f0 --- /dev/null +++ b/src/pybind/mgr/rook/rook_client/_helper.py @@ -0,0 +1,108 @@ +import logging +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 _property_from_json(data, breadcrumb, name, py_name, typ, required, nullable): + if not required and name not in data: + return _omit + obj = data[name] + if nullable and obj is None: + return obj + 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, required, nullable): + obj = getattr(self, '_' + py_name) + 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(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 + return [e.to_json() for e in self] + + @classmethod + def from_json(cls, data, breadcrumb=''): + if cls._items_type is None: + return cls(data) + return cls(cls._items_type.from_json(e, breadcrumb + '[]') for e in data) |