ceph/src/pybind/mgr/rook/rook-client-python/generate_model_classes.py

   1 """
   2 Generate Python files containing data Python models classes for
   3 all properties of the all CRDs in the file
   4
   5 **Note**: generate_model_classes.py is independent of Rook or Ceph. It can be used for all
   6   CRDs.
   7
   8 For example:
   9   python3 -m venv venv
  10   pip install -r requirements.txt
  11   python generate_model_classes.py <crds.yaml> <output-folder>
  12   python setup.py develop
  13
  14 Usage:
  15   generate_model_classes.py <crds.yaml> <output-folder>
  16 """
  17 import os
  18 from abc import ABC, abstractmethod
  19 from collections import OrderedDict
  20 from typing import List, Union, Iterator, Optional
  21
  22 import yaml
  23 try:
  24     from dataclasses import dataclass
  25 except ImportError:
  26     from attr import dataclass  # type: ignore
  27
  28 header = '''"""
  29 This file is automatically generated.
  30 Do not modify.
  31 """
  32
  33 try:
  34     from typing import Any, Optional, Union, List
  35 except ImportError:
  36     pass
  37
  38 from .._helper import _omit, CrdObject, CrdObjectList, CrdClass
  39
  40 '''
  41
  42 @dataclass  # type: ignore
  43 class CRDBase(ABC):
  44     name: str
  45     nullable: bool
  46     required: bool
  47
  48     @property
  49     def py_name(self):
  50         return self.name
  51
  52     @property
  53     @abstractmethod
  54     def py_type(self):
  55         ...
  56
  57     @abstractmethod
  58     def flatten(self):
  59         ...
  60
  61     def py_property(self):
  62         return f"""
  63 @property
  64 def {self.py_name}(self):
  65     # type: () -> {self.py_property_return_type}
  66     return self._property_impl('{self.py_name}')
  67
  68 @{self.py_name}.setter
  69 def {self.py_name}(self, new_val):
  70     # type: ({self.py_param_type}) -> None
  71     self._{self.py_name} = new_val
  72     """.strip()
  73
  74     @property
  75     def py_param(self):
  76         if not self.has_default:
  77             return f'{self.py_name},  # type: {self.py_param_type}'
  78         return f'{self.py_name}=_omit,  # type: {self.py_param_type}'
  79
  80     @property
  81     def has_default(self):
  82         return not self.required
  83
  84     @property
  85     def py_param_type(self):
  86         return f'Optional[{self.py_type}]' if (self.nullable or not self.required) else self.py_type
  87
  88     @property
  89     def py_property_return_type(self):
  90         return f'Optional[{self.py_type}]' if (self.nullable) else self.py_type
  91
  92 @dataclass
  93 class CRDAttribute(CRDBase):
  94     type: str
  95     default_value: str='_omit'
  96
  97     @property
  98     def py_param(self):
  99         if not self.has_default:
 100             return f'{self.py_name},  # type: {self.py_param_type}'
 101         return f'{self.py_name}={self.default_value},  # type: {self.py_param_type}'
 102
 103     @property
 104     def has_default(self):
 105         return not self.required or self.default_value != '_omit'
 106
 107     @property
 108     def py_type(self):
 109         return {
 110             'integer': 'int',
 111             'boolean': 'bool',
 112             'string': 'str',
 113             'object': 'Any',
 114             'number': 'float',
 115         }[self.type]
 116
 117     def flatten(self):
 118         yield from ()
 119
 120     def toplevel(self):
 121         return ''
 122
 123
 124 @dataclass
 125 class CRDList(CRDBase):
 126     items: 'CRDClass'
 127
 128     @property
 129     def py_name(self):
 130         return self.name
 131
 132     @property
 133     def py_type(self):
 134         return self.name[0].upper() + self.name[1:] + 'List'
 135
 136     @property
 137     def py_param_type(self):
 138         inner = f'Union[List[{self.items.py_type}], CrdObjectList]'
 139         return f'Optional[{inner}]' if (self.nullable or not self.required) else inner
 140
 141     @property
 142     def py_property_return_type(self):
 143         inner = f'Union[List[{self.items.py_type}], CrdObjectList]'
 144         return f'Optional[{inner}]' if (self.nullable) else inner
 145
 146     def flatten(self):
 147         yield from self.items.flatten()
 148         yield self
 149
 150     def toplevel(self):
 151         py_type = self.items.py_type
 152         if py_type == 'Any':
 153             py_type = 'None'
 154
 155         return f"""
 156 class {self.py_type}(CrdObjectList):
 157 {indent('_items_type = ' + py_type)}
 158 """.strip()
 159
 160
 161 @dataclass
 162 class CRDClass(CRDBase):
 163     attrs: List[Union[CRDAttribute, 'CRDClass']]
 164     base_class: str = 'CrdObject'
 165
 166     def toplevel(self):
 167         ps = '\n\n'.join(a.py_property() for a in self.attrs)
 168         return f"""class {self.py_type}({self.base_class}):
 169 {indent(self.py_properties())}
 170
 171 {indent(self.py_init())}
 172
 173 {indent(ps)}
 174 """.strip()
 175
 176     @property
 177     def sub_classes(self) -> List["CRDClass"]:
 178         return [a for a in self.attrs if isinstance(a, CRDClass)]
 179
 180     @property
 181     def py_type(self):
 182         return self.name[0].upper() + self.name[1:]
 183
 184     def py_properties(self):
 185         def a_to_tuple(a):
 186             return ', '.join((f"'{a.name}'",
 187                        f"'{a.py_name}'",
 188                        a.py_type.replace('Any', 'object'),
 189                        str(a.required),
 190                        str(a.nullable)))
 191
 192         attrlist = ',\n'.join([f'({a_to_tuple(a)})' for a in self.attrs])
 193         return f"""_properties = [\n{indent(attrlist)}\n]"""
 194
 195     def flatten(self) -> Iterator['CRDClass']:
 196         for sub_cls in self.attrs:
 197             yield from sub_cls.flatten()
 198         yield self
 199
 200     def py_init(self):
 201         sorted_attrs = sorted(self.attrs, key=lambda a: a.has_default)
 202         params = '\n'.join(a.py_param for a in sorted_attrs)
 203         params_set = '\n'.join(f'{a.py_name}={a.py_name},' for a in sorted_attrs)
 204         return f"""
 205 def __init__(self,
 206 {indent(params, indent=4+9)}
 207              ):
 208     super({self.py_type}, self).__init__(
 209 {indent(params_set, indent=8)}
 210     )
 211 """.strip()
 212
 213 def indent(s, indent=4):
 214     return '\n'.join(' '*indent + l for l in s.splitlines())
 215
 216
 217 def handle_property(elem_name, elem: dict, required: bool):
 218     nullable = elem.get('nullable', False)
 219     if 'properties' in elem:
 220         ps = elem['properties']
 221         required_elems = elem.get('required', [])
 222         sub_props = [handle_property(k, v, k in required_elems) for k, v in ps.items()]
 223         return CRDClass(elem_name, nullable, required, sub_props)
 224     elif 'items' in elem:
 225         item = handle_property(elem_name + 'Item', elem['items'], False)
 226         return CRDList(elem_name, nullable, required, item)
 227     elif 'type' in elem:
 228         return CRDAttribute(elem_name, nullable, required, elem['type'])
 229     elif elem == {}:
 230         return CRDAttribute(elem_name, nullable, required, 'object')
 231     assert False, str((elem_name, elem))
 232
 233
 234 def handle_crd(c_dict) -> Optional[CRDClass]:
 235     try:
 236         name = c_dict['spec']['names']['kind']
 237         s = c_dict['spec']['validation']['openAPIV3Schema']
 238     except (KeyError, TypeError):
 239         return None
 240     s['required'] = ['spec']
 241     c = handle_property(name, s, True)
 242     k8s_attrs = [CRDAttribute('apiVersion', False, True, 'string'),
 243                  CRDAttribute('metadata', False, True, 'object'),
 244                  CRDAttribute('status', False, False, 'object')]
 245     return CRDClass(c.name, False, True, k8s_attrs + c.attrs, base_class='CrdClass')
 246
 247
 248 def local(yaml_filename):
 249     with open(yaml_filename) as f:
 250         yamls = yaml.safe_load_all(f.read())
 251         for y in yamls:
 252             try:
 253                 yield y
 254             except AttributeError:
 255                 pass
 256
 257
 258 def remove_duplicates(items):
 259     return OrderedDict.fromkeys(items).keys()
 260
 261
 262 def get_toplevels(crd):
 263     elems = list(crd.flatten())
 264
 265     def dup_elems(l):
 266         ds = set([x for x in l if l.count(x) > 1])
 267         return ds
 268
 269     names = [t.name for t in elems]
 270     for dup_name in dup_elems(names):
 271         dups = set(e.toplevel() for e in elems if e.name == dup_name)
 272         assert len(dups) == 1, str(dups)
 273
 274     return remove_duplicates(cls.toplevel() for cls in elems)
 275
 276
 277 def main(yaml_filename, outfolder):
 278     for crd in local(yaml_filename):
 279         valid_crd = handle_crd(crd)
 280         if valid_crd is not None:
 281             try:
 282                 os.mkdir(outfolder)
 283             except FileExistsError:
 284                 pass
 285             open(f'{outfolder}/__init__.py', 'w').close()
 286
 287             with open(f'{outfolder}/{valid_crd.name.lower()}.py', 'w') as f:
 288                 f.write(header)
 289                 classes = get_toplevels(valid_crd)
 290                 f.write('\n\n\n'.join(classes))
 291                 f.write('\n')
 292
 293
 294 if __name__ == '__main__':
 295     from docopt import docopt
 296     args = docopt(__doc__)
 297     yaml_filename = '/dev/stdin' if args["<crds.yaml>"] == '-' else args["<crds.yaml>"]
 298     main(yaml_filename, args["<output-folder>"])