]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/rook/rook-client-python/rook_client/_helper.py
4 from typing
import List
, Dict
, Any
, Optional
8 logger
= logging
.getLogger(__name__
)
10 # Tricking mypy to think `_omit`'s type is NoneType
11 # To make us not add things like `Union[Optional[str], OmitType]`
13 _omit
= None # type: NoneType
14 _omit
= object() # type: ignore
17 # Don't add any additionalProperties to objects. Useful for completeness testing
20 def _str_to_class(cls
, typ_str
):
21 if isinstance(typ_str
, str):
22 return getattr(sys
.modules
[cls
.__module
__], typ_str
)
26 def _property_from_json(cls
, data
, breadcrumb
, name
, py_name
, typ_str
, required
, nullable
):
27 if not required
and name
not in data
:
32 raise ValueError('KeyError in {}: {}'.format(breadcrumb
, e
))
33 if nullable
and obj
is None:
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
)
41 class CrdObject(object):
42 _properties
= [] # type: List
44 def __init__(self
, **kwargs
):
45 for prop
in self
._properties
:
46 setattr(self
, prop
[1], kwargs
.pop(prop
[1]))
49 '{} got unexpected arguments {}'.format(self
.__class
__.__name
__, kwargs
.keys()))
50 self
._additionalProperties
= {} # type: Dict[str, Any]
52 def _property_impl(self
, name
):
53 obj
= getattr(self
, '_' + name
)
55 raise AttributeError(name
+ ' not found')
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:
64 if not required
and obj
is _omit
:
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
}
77 def from_json(cls
, data
, breadcrumb
=''):
80 p
[1]: _property_from_json(cls
, data
, breadcrumb
, *p
) for p
in cls
._properties
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
86 except (TypeError, AttributeError, KeyError):
87 logger
.exception(breadcrumb
)
91 class CrdClass(CrdObject
):
93 def from_json(cls
, data
, breadcrumb
=''):
95 if kind
!= cls
.__name
__:
96 raise ValueError("kind mismatch: {} != {}".format(kind
, cls
.__name
__))
97 return super(CrdClass
, cls
).from_json(data
, breadcrumb
)
100 ret
= super(CrdClass
, self
).to_json()
101 ret
['kind'] = self
.__class
__.__name
__
105 class CrdObjectList(list):
106 # Py3: Replace `Any` with `TypeVar('T_CrdObject', bound='CrdObject')`
107 _items_type
= None # type: Optional[Any]
111 if self
._items
_type
is None:
113 if issubclass(self
._items
_type
, CrdObject
) or issubclass(self
._items
_type
, CrdObjectList
):
114 return [e
.to_json() for e
in self
]
119 def from_json(cls
, data
, breadcrumb
=''):
120 if cls
._items
_type
is None:
122 if issubclass(cls
._items
_type
, CrdObject
) or issubclass(cls
._items
_type
, CrdObjectList
):
123 return cls(cls
._items
_type
.from_json(e
, breadcrumb
+ '[{}]'.format(i
)) for i
, e
in enumerate(data
))
127 return '{}({})'.format(self
.__class
__.__name
__, repr(list(self
)))