ceph/src/pybind/mgr/rook/rook-client-python/rook_client/_helper.py.orig

   1 import logging
   2 import sys
   3 try:
   4     from typing import List, Dict, Any, Optional
   5 except ImportError:
   6     pass
   7
   8 logger = logging.getLogger(__name__)
   9
  10 # Tricking mypy to think `_omit`'s type is NoneType
  11 # To make us not add things like `Union[Optional[str], OmitType]`
  12 NoneType = type(None)
  13 _omit = None  # type: NoneType
  14 _omit = object()  # type: ignore
  15
  16
  17 # Don't add any additionalProperties to objects. Useful for completeness testing
  18 STRICT = False
  19
  20 def _str_to_class(cls, typ_str):
  21     if isinstance(typ_str, str):
  22         return getattr(sys.modules[cls.__module__], typ_str)
  23     return typ_str
  24
  25
  26 def _property_from_json(cls, data, breadcrumb, name, py_name, typ_str, required, nullable):
  27     if not required and name not in data:
  28         return _omit
  29     try:
  30         obj = data[name]
  31     except KeyError as e:
  32         raise ValueError('KeyError in {}: {}'.format(breadcrumb, e))
  33     if nullable and obj is None:
  34         return obj
  35     typ = _str_to_class(cls, typ_str)
  36     if issubclass(typ, CrdObject) or issubclass(typ, CrdObjectList):
  37         return typ.from_json(obj, breadcrumb + '.' + name)
  38     return obj
  39
  40
  41 class CrdObject(object):
  42     _properties = []  # type: List
  43
  44     def __init__(self, **kwargs):
  45         for prop in self._properties:
  46             setattr(self, prop[1], kwargs.pop(prop[1]))
  47         if kwargs:
  48             raise TypeError(
  49                 '{} got unexpected arguments {}'.format(self.__class__.__name__, kwargs.keys()))
  50         self._additionalProperties = {}  # type: Dict[str, Any]
  51
  52     def _property_impl(self, name):
  53         obj = getattr(self, '_' + name)
  54         if obj is _omit:
  55             raise AttributeError(name + ' not found')
  56         return obj
  57
  58     def _property_to_json(self, name, py_name, typ_str, required, nullable):
  59         obj = getattr(self, '_' + py_name)
  60         typ = _str_to_class(self.__class__, typ_str)
  61         if issubclass(typ, CrdObject) or issubclass(typ, CrdObjectList):
  62             if nullable and obj is None:
  63                 return obj
  64             if not required and obj is _omit:
  65                 return obj
  66             return obj.to_json()
  67         else:
  68             return obj
  69
  70     def to_json(self):
  71         # type: () -> Dict[str, Any]
  72         res = {p[0]: self._property_to_json(*p) for p in self._properties}
  73         res.update(self._additionalProperties)
  74         return {k: v for k, v in res.items() if v is not _omit}
  75
  76     @classmethod
  77     def from_json(cls, data, breadcrumb=''):
  78         try:
  79             sanitized = {
  80                 p[1]: _property_from_json(cls, data, breadcrumb, *p) for p in cls._properties
  81             }
  82             extra = {k:v for k,v in data.items() if k not in sanitized}
  83             ret = cls(**sanitized)
  84             ret._additionalProperties = {} if STRICT else extra
  85             return ret
  86         except (TypeError, AttributeError, KeyError):
  87             logger.exception(breadcrumb)
  88             raise
  89
  90
  91 class CrdClass(CrdObject):
  92     @classmethod
  93     def from_json(cls, data, breadcrumb=''):
  94         kind = data['kind']
  95         if kind != cls.__name__:
  96             raise ValueError("kind mismatch: {} != {}".format(kind, cls.__name__))
  97         return super(CrdClass, cls).from_json(data, breadcrumb)
  98
  99     def to_json(self):
 100         ret = super(CrdClass, self).to_json()
 101         ret['kind'] = self.__class__.__name__
 102         return ret
 103
 104
 105 class CrdObjectList(list):
 106     # Py3: Replace `Any` with `TypeVar('T_CrdObject', bound='CrdObject')`
 107     _items_type = None  # type: Optional[Any]
 108
 109     def to_json(self):
 110         # type: () -> List
 111         if self._items_type is None:
 112             return self
 113         if issubclass(self._items_type, CrdObject) or issubclass(self._items_type, CrdObjectList):
 114             return [e.to_json() for e in self]
 115         return list(self)
 116
 117
 118     @classmethod
 119     def from_json(cls, data, breadcrumb=''):
 120         if cls._items_type is None:
 121             return cls(data)
 122         if issubclass(cls._items_type, CrdObject) or issubclass(cls._items_type, CrdObjectList):
 123 <<<<<<< HEAD
 124             return cls(cls._items_type.from_json(e, breadcrumb + '[]') for e in data)
 125         return cls(data)
 126
 127     def __repr__(self):
 128         return '{}({})'.format(self.__class__.__name__, repr(list(self)))
 129 =======
 130             return cls(cls._items_type.from_json(e, breadcrumb + '[{}]'.format(i)) for i, e in enumerate(data))
 131         return data
 132 >>>>>>> 2e4a0b5 (Better ex message for missing required elements)
 133