Compare commits

...

4 Commits

Author SHA1 Message Date
chacha
4f5dade949 first feature implementation 2025-09-08 01:23:46 +02:00
cclecle
cce260bc5e reordering 2025-09-07 18:42:38 +02:00
cclecle
915a4332ee tiny fix :) 2025-09-06 01:47:49 +02:00
cclecle
4dca3eb9d1 improve typing 2025-09-06 01:43:20 +02:00
3 changed files with 152 additions and 78 deletions

View File

@@ -27,4 +27,5 @@ from .model import (
IncompletelyAnnotatedField,
ImportForbidden,
FunctionForbidden,
FrozenDABField
)

View File

@@ -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.

View File

@@ -1175,7 +1175,24 @@ class TestConfigWithoutEnabledFlag(unittest.TestCase):
def __initializer(cls):
test_initializer_safe_testfc()
def test_feature(self):
"""Testing first appliance level, and Field types (simple)"""
# class can be created
class Appliance1(dm.BaseAppliance):
VarStrOuter: str = "testvalue APPLIANCE"
class Feature1(dm.BaseFeature):
VarStrInner: str = "testvalue FEATURE"
app1 = Appliance1()
self.assertIsInstance(app1.__DABSchema__["VarStrOuter"],dm.FrozenDABField)
self.assertTrue(hasattr(app1, "Feature1"))
self.assertTrue(hasattr(app1.Feature1, "VarStrInner"))
self.assertIn("Feature1",app1.__DABFeatures__)
self.assertIn("VarStrInner",app1.__DABFeatures__["Feature1"].__DABSchema__)
self.assertIsInstance(app1.__DABFeatures__["Feature1"].__DABSchema__["VarStrInner"],dm.FrozenDABField)
# ---------- main ----------
if __name__ == "__main__":