|
|
|
|
@@ -55,7 +55,6 @@ ALLOWED_ANNOTATIONS = {
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
ALLOWED_MODEL_FIELDS_TYPES = (str, int, float, complex, bool, bytes)
|
|
|
|
|
ALLOWED_MODEL_FIELDS_CONTAINERS = (dict, list, set, frozenset, tuple)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# TV_ALLOWED_MODEL_FIELDS_TYPES = TypeVar("TV_ALLOWED_MODEL_FIELDS_TYPES", *ALLOWED_MODEL_FIELDS_TYPES, *ALLOWED_MODEL_FIELDS_CONTAINERS)
|
|
|
|
|
@@ -66,7 +65,11 @@ class DABModelException(Exception):
|
|
|
|
|
Base Exception for DABModelException class
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class ReservedFieldName(Exception):
|
|
|
|
|
"""DABModelException Exception class
|
|
|
|
|
Base Exception for DABModelException class
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
class MultipleInheritanceForbidden(DABModelException):
|
|
|
|
|
"""MultipleInheritanceForbidden Exception class
|
|
|
|
|
Multiple inheritance is forbidden when using dabmodel
|
|
|
|
|
@@ -298,7 +301,7 @@ class DABFieldInfo:
|
|
|
|
|
self._constraints = constraints
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def doc(self):
|
|
|
|
|
def doc(self) -> str:
|
|
|
|
|
"""Returns Field's documentation"""
|
|
|
|
|
return self._doc
|
|
|
|
|
|
|
|
|
|
@@ -325,7 +328,7 @@ class DABField(Generic[T_Field]):
|
|
|
|
|
self._source = s
|
|
|
|
|
|
|
|
|
|
@property
|
|
|
|
|
def doc(self):
|
|
|
|
|
def doc(self) -> str:
|
|
|
|
|
"""Returns Field's documentation"""
|
|
|
|
|
return self._info.doc
|
|
|
|
|
|
|
|
|
|
@@ -403,10 +406,14 @@ class ModelSpecView:
|
|
|
|
|
|
|
|
|
|
__slots__ = ("_vals", "_types", "_touched", "_name", "_module")
|
|
|
|
|
|
|
|
|
|
def __init__(self, values: dict, types_map: dict[str, type], name, module):
|
|
|
|
|
def __init__(self, values: dict[str, Any], types_map: dict[str, type], name: str, module: str):
|
|
|
|
|
self._name: str
|
|
|
|
|
self._vals: dict[str, Any]
|
|
|
|
|
self._types: dict[str, type]
|
|
|
|
|
self._touched: set
|
|
|
|
|
self._module: str
|
|
|
|
|
object.__setattr__(self, "_vals", dict(values))
|
|
|
|
|
object.__setattr__(self, "_types", types_map)
|
|
|
|
|
object.__setattr__(self, "_touched", set())
|
|
|
|
|
object.__setattr__(self, "_name", name)
|
|
|
|
|
object.__setattr__(self, "_module", module)
|
|
|
|
|
|
|
|
|
|
@@ -421,16 +428,16 @@ class ModelSpecView:
|
|
|
|
|
return self._module
|
|
|
|
|
|
|
|
|
|
@__module__.setter
|
|
|
|
|
def __module__(self, value):
|
|
|
|
|
def __module__(self, value: str):
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
def __getattr__(self, name):
|
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
|
|
|
"""internal proxy getattr"""
|
|
|
|
|
if name not in self._types:
|
|
|
|
|
raise AttributeError(f"Unknown field {name}")
|
|
|
|
|
return self._vals[name]
|
|
|
|
|
|
|
|
|
|
def __setattr__(self, name, value):
|
|
|
|
|
def __setattr__(self, name: str, value: Any):
|
|
|
|
|
"""internal proxy setattr"""
|
|
|
|
|
if name not in self._types:
|
|
|
|
|
raise NonExistingField(f"Cannot set unknown field {name}")
|
|
|
|
|
@@ -446,7 +453,6 @@ class ModelSpecView:
|
|
|
|
|
raise InvalidFieldValue(f"Field <{name}> value is not of expected type {T}.") from exp
|
|
|
|
|
|
|
|
|
|
self._vals[name] = value
|
|
|
|
|
self._touched.add(name)
|
|
|
|
|
|
|
|
|
|
def export(self) -> dict:
|
|
|
|
|
"""exports all proxified values"""
|
|
|
|
|
@@ -489,7 +495,40 @@ class BaseMeta(type):
|
|
|
|
|
namespace[_funknown] = None
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def pre_processing_modified(
|
|
|
|
|
def pre_processing(mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any]):
|
|
|
|
|
"""preprocessing BaseElement"""
|
|
|
|
|
# iterating new and modified fields
|
|
|
|
|
mcs.modified_field = {}
|
|
|
|
|
mcs.new_fields = {}
|
|
|
|
|
mcs.initializer = None
|
|
|
|
|
initializer_name: Optional[str] = None
|
|
|
|
|
for _fname, _fvalue in namespace.items():
|
|
|
|
|
if _fname == f"_{name}__initializer" or (name.startswith("_") and _fname == "__initializer"):
|
|
|
|
|
if not isinstance(_fvalue, classmethod):
|
|
|
|
|
raise InvalidInitializerType()
|
|
|
|
|
mcs.initializer = _fvalue.__func__
|
|
|
|
|
if name.startswith("_"):
|
|
|
|
|
initializer_name = "__initializer"
|
|
|
|
|
else:
|
|
|
|
|
initializer_name = f"_{name}__initializer"
|
|
|
|
|
elif _fname.startswith("__"):
|
|
|
|
|
pass
|
|
|
|
|
else:
|
|
|
|
|
# print(f"Parsing Field: {_fname} / {_fvalue}")
|
|
|
|
|
if len(bases) == 1 and _fname in namespace["__DABSchema__"].keys(): # Modified fields
|
|
|
|
|
mcs.pre_processing_modified_fields(name, bases, namespace, _fname, _fvalue)
|
|
|
|
|
else: # New fieds
|
|
|
|
|
mcs.pre_processing_new_fields(name, bases, namespace, _fname, _fvalue)
|
|
|
|
|
# removing modified fields from class (will add them back later)
|
|
|
|
|
for _fname in mcs.new_fields:
|
|
|
|
|
del namespace[_fname]
|
|
|
|
|
for _fname in mcs.modified_field:
|
|
|
|
|
del namespace[_fname]
|
|
|
|
|
if mcs.initializer is not None and initializer_name is not None:
|
|
|
|
|
del namespace[initializer_name]
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def pre_processing_modified_fields(
|
|
|
|
|
mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], _fname: str, _fvalue: Any
|
|
|
|
|
): # pylint: disable=unused-argument
|
|
|
|
|
"""preprocessing BaseElement modified Fields"""
|
|
|
|
|
@@ -509,7 +548,7 @@ class BaseMeta(type):
|
|
|
|
|
mcs.modified_field[_fname] = _fvalue
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def pre_processing_new(
|
|
|
|
|
def pre_processing_new_fields(
|
|
|
|
|
mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], _fname: str, _fvalue: Any
|
|
|
|
|
): # pylint: disable=unused-argument
|
|
|
|
|
"""preprocessing BaseElement new Fields"""
|
|
|
|
|
@@ -549,43 +588,10 @@ class BaseMeta(type):
|
|
|
|
|
raise InvalidFieldValue(f"Value of Field <{_fname}> is not of expected type {namespace['__annotations__'][_fname]}.") from exp
|
|
|
|
|
mcs.new_fields[_fname] = DABField(_fname, _fvalue, namespace["__annotations__"][_fname], _finfo)
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def pre_processing(mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any]):
|
|
|
|
|
"""preprocessing BaseElement"""
|
|
|
|
|
# iterating new and modified fields
|
|
|
|
|
mcs.modified_field = {}
|
|
|
|
|
mcs.new_fields = {}
|
|
|
|
|
mcs.initializer = None
|
|
|
|
|
initializer_name: Optional[str] = None
|
|
|
|
|
for _fname, _fvalue in namespace.items():
|
|
|
|
|
if _fname == f"_{name}__initializer" or (name.startswith("_") and _fname == "__initializer"):
|
|
|
|
|
if not isinstance(_fvalue, classmethod):
|
|
|
|
|
raise InvalidInitializerType()
|
|
|
|
|
mcs.initializer = _fvalue.__func__
|
|
|
|
|
if name.startswith("_"):
|
|
|
|
|
initializer_name = "__initializer"
|
|
|
|
|
else:
|
|
|
|
|
initializer_name = f"_{name}__initializer"
|
|
|
|
|
elif _fname.startswith("__"):
|
|
|
|
|
pass
|
|
|
|
|
else:
|
|
|
|
|
# print(f"Parsing Field: {_fname} / {_fvalue}")
|
|
|
|
|
if len(bases) == 1 and _fname in namespace["__DABSchema__"].keys(): # Modified fields
|
|
|
|
|
mcs.pre_processing_modified(name, bases, namespace, _fname, _fvalue)
|
|
|
|
|
else: # New fieds
|
|
|
|
|
mcs.pre_processing_new(name, bases, namespace, _fname, _fvalue)
|
|
|
|
|
# removing modified fields from class (will add them back later)
|
|
|
|
|
for _fname in mcs.new_fields:
|
|
|
|
|
del namespace[_fname]
|
|
|
|
|
for _fname in mcs.modified_field:
|
|
|
|
|
del namespace[_fname]
|
|
|
|
|
if mcs.initializer is not None and initializer_name is not None:
|
|
|
|
|
del namespace[initializer_name]
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def call_initializer(
|
|
|
|
|
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any]
|
|
|
|
|
): # pylint: disable=unused-argument
|
|
|
|
|
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any] # pylint: disable=unused-argument
|
|
|
|
|
):
|
|
|
|
|
"""BaseElement initializer processing"""
|
|
|
|
|
if mcs.initializer is not None:
|
|
|
|
|
init_fieldvalues = {}
|
|
|
|
|
@@ -613,69 +619,119 @@ class BaseMeta(type):
|
|
|
|
|
f"Value of Field <{_fname}> is not of expected type {namespace['__annotations__'][_fname]}."
|
|
|
|
|
) from exp
|
|
|
|
|
cls.__DABSchema__[_fname].update_value(_fvalue)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def __new__(mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any]) -> Type:
|
|
|
|
|
"""BaseElement new class"""
|
|
|
|
|
mcs.pre_check(name, bases, namespace)
|
|
|
|
|
mcs.pre_processing(name, bases, namespace)
|
|
|
|
|
|
|
|
|
|
orig_setattr = namespace.get("__setattr__", object.__setattr__)
|
|
|
|
|
|
|
|
|
|
def guarded_setattr(self, key, value):
|
|
|
|
|
if key.startswith("_"): # allow private and dunder attrs
|
|
|
|
|
return orig_setattr(self, key, value)
|
|
|
|
|
# block writes after init if key is readonly
|
|
|
|
|
if key in self.__DABSchema__.keys():
|
|
|
|
|
if hasattr(self, key):
|
|
|
|
|
raise ReadOnlyField(f"{key} is read-only")
|
|
|
|
|
else:
|
|
|
|
|
raise NewFieldForbidden("creating new fields is not allowed")
|
|
|
|
|
|
|
|
|
|
return orig_setattr(self, key, value)
|
|
|
|
|
|
|
|
|
|
namespace["__setattr__"] = guarded_setattr
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
_cls = super().__new__(mcs, name, bases, namespace)
|
|
|
|
|
|
|
|
|
|
for _fname, _fvalue in mcs.modified_field.items():
|
|
|
|
|
_cls.__DABSchema__[_fname] = deepcopy(bases[0].__DABSchema__[_fname])
|
|
|
|
|
_cls.__DABSchema__[_fname].update_value(_fvalue)
|
|
|
|
|
|
|
|
|
|
for _fname, _fvalue in mcs.new_fields.items():
|
|
|
|
|
_fvalue.add_source(mcs)
|
|
|
|
|
_cls.__DABSchema__[_fname] = _fvalue
|
|
|
|
|
|
|
|
|
|
mcs.save_values(_cls, name, bases, namespace)
|
|
|
|
|
mcs.call_initializer(_cls, name, bases, namespace)
|
|
|
|
|
|
|
|
|
|
return _cls
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def save_values(
|
|
|
|
|
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any] # pylint: disable=unused-argument
|
|
|
|
|
):
|
|
|
|
|
for _fname, _fvalue in mcs.modified_field.items():
|
|
|
|
|
cls.__DABSchema__[_fname] = deepcopy(bases[0].__DABSchema__[_fname])
|
|
|
|
|
cls.__DABSchema__[_fname].update_value(_fvalue)
|
|
|
|
|
|
|
|
|
|
for _fname, _fvalue in mcs.new_fields.items():
|
|
|
|
|
_fvalue.add_source(mcs)
|
|
|
|
|
cls.__DABSchema__[_fname] = _fvalue
|
|
|
|
|
|
|
|
|
|
def __call__(cls, *args: Any, **kw: Any): # intentionally untyped
|
|
|
|
|
"""BaseElement new instance"""
|
|
|
|
|
obj = super().__call__(*args, **kw)
|
|
|
|
|
|
|
|
|
|
for _fname, _fvalue in cls.__DABSchema__.items():
|
|
|
|
|
setattr(obj, _fname, _fvalue)
|
|
|
|
|
setattr(obj, _fname, _fvalue.value)
|
|
|
|
|
inst_schema: dict[str, Any] = {}
|
|
|
|
|
for _fname, _fvalue in cls.__DABSchema__.items():
|
|
|
|
|
inst_schema[_fname] = FrozenDABField(_fvalue)
|
|
|
|
|
#print(inst_schema)
|
|
|
|
|
setattr(obj, "__DABSchema__", inst_schema)
|
|
|
|
|
return obj
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
cls.modify_object(obj)
|
|
|
|
|
cls.install_guard(obj)
|
|
|
|
|
return obj
|
|
|
|
|
|
|
|
|
|
def modify_object(self,obj):
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
def install_guard(self, obj):
|
|
|
|
|
orig_setattr = getattr(self,"__setattr__")
|
|
|
|
|
|
|
|
|
|
def guarded_setattr(_self, key: str, value: Any):
|
|
|
|
|
if key.startswith("_"): # allow private and dunder attrs
|
|
|
|
|
return orig_setattr(_self, key, value)
|
|
|
|
|
# block writes after init if key is readonly
|
|
|
|
|
if key in _self.__DABSchema__.keys():
|
|
|
|
|
if hasattr(_self, key):
|
|
|
|
|
raise ReadOnlyField(f"{key} is read-only")
|
|
|
|
|
else:
|
|
|
|
|
raise NewFieldForbidden("creating new fields is not allowed")
|
|
|
|
|
|
|
|
|
|
return orig_setattr(_self, key, value)
|
|
|
|
|
|
|
|
|
|
setattr(self, "__setattr__", guarded_setattr)
|
|
|
|
|
|
|
|
|
|
class BaseElement(metaclass=BaseMeta):
|
|
|
|
|
"""BaseElement class
|
|
|
|
|
Base class to apply metaclass and set common Fields.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
class BaseMetaFeature(BaseMeta):
|
|
|
|
|
pass
|
|
|
|
|
|
|
|
|
|
class BaseFeature(BaseElement):
|
|
|
|
|
class BaseFeature(BaseElement,metaclass=BaseMetaFeature):
|
|
|
|
|
"""BaseFeature class
|
|
|
|
|
Base class for Appliance's Features.
|
|
|
|
|
Features are optional traits of an appliance.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class BaseAppliance(BaseElement):
|
|
|
|
|
class BaseMetaAppliance(BaseMeta):
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def pre_processing(mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any]):
|
|
|
|
|
mcs.features: dict[str,type[BaseFeature]] = {}
|
|
|
|
|
super().pre_processing(name,bases,namespace)
|
|
|
|
|
for _ftname in mcs.features.keys():
|
|
|
|
|
del namespace[_ftname]
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def pre_processing_new_fields(
|
|
|
|
|
mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], _fname: str, _fvalue: Any
|
|
|
|
|
): # pylint: disable=unused-argument
|
|
|
|
|
"""preprocessing BaseElement new Fields"""
|
|
|
|
|
if _fname == "features":
|
|
|
|
|
raise ReservedFieldName("feature is a reserved field name")
|
|
|
|
|
elif _fname =="get_features":
|
|
|
|
|
pass
|
|
|
|
|
elif isinstance(_fvalue,BaseMetaFeature):
|
|
|
|
|
mcs.features[_fname]=_fvalue
|
|
|
|
|
else:
|
|
|
|
|
super().pre_processing_new_fields(name,bases,namespace,_fname,_fvalue)
|
|
|
|
|
|
|
|
|
|
@classmethod
|
|
|
|
|
def save_values(
|
|
|
|
|
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any] # pylint: disable=unused-argument
|
|
|
|
|
):
|
|
|
|
|
super().save_values(cls,name,bases,namespace)
|
|
|
|
|
cls.__DABFeatures__ = mcs.features
|
|
|
|
|
|
|
|
|
|
def modify_object(self, obj): # intentionally untyped
|
|
|
|
|
for _ftname,_ftvalue in self.features.items():
|
|
|
|
|
instft = _ftvalue()
|
|
|
|
|
setattr(self, _ftname,instft )
|
|
|
|
|
self.__DABFeatures__[_ftname] = instft
|
|
|
|
|
|
|
|
|
|
class BaseAppliance(BaseElement,metaclass=BaseMetaAppliance):
|
|
|
|
|
"""BaseFeature class
|
|
|
|
|
Base class for Appliance.
|
|
|
|
|
An appliance is a server configuration / image that is built using appliance's code and Fields.
|
|
|
|
|
|