chacha/dabmodel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 45 Wiki Activity

reorganize and rename (partial)

Browse Source
This commit is contained in:
chacha
2025-09-20 18:27:36 +02:00
parent af81ec5fd3
commit 9aec2d66cd
14 changed files with 639 additions and 586 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
src/dabmodel/LAMFields/Constraint.py Normal file
View File

@@ -0,0 +1,17 @@
from typing import Generic, TypeVar
T_Field = TypeVar("T_Field")
class BaseConstraint(Generic[T_Field]):
"""BaseConstraint class
Base class for Field's constraints
"""
_bound_type: type
def __init__(self): ...
def check(self, value: T_Field) -> bool:
"""Check if a Constraint is completed"""
return True

41
src/dabmodel/LAMFields/FrozenLAMField.py Normal file
View File

@@ -0,0 +1,41 @@
from typing import Generic, TypeVar, Any
from .LAMField import LAMField
from .Constraint import BaseConstraint
from ..tools import LAMdeepfreeze
T_Field = TypeVar("T_Field")
class FrozenLAMField(Generic[T_Field]):
"""FrozenLAMField class
a read-only proxy of a Field
"""
def __init__(self, inner_field: LAMField):
self._inner_field = inner_field
@property
def doc(self) -> str:
"""Returns Field's documentation (frozen)"""
return LAMdeepfreeze(self._inner_field.doc)
@property
def constraints(self) -> tuple[BaseConstraint]:
"""Returns Field's constraint (frozen)"""
return LAMdeepfreeze(self._inner_field.constraints)
@property
def default_value(self) -> Any:
"""Returns Field's default value (frozen)"""
return self._inner_field.default_value
@property
def value(self) -> Any:
"""Returns Field's value (frosen)"""
return self._inner_field.value
@property
def annotations(self) -> Any:
"""Returns Field's annotation (frozen)"""
return LAMdeepfreeze(self._inner_field.annotations)

62
src/dabmodel/LAMFields/LAMField.py Normal file
View File

@@ -0,0 +1,62 @@
from typing import Generic, TypeVar, Optional, Any
from .LAMFieldInfo import LAMFieldInfo
from .Constraint import BaseConstraint
from ..tools import LAMdeepfreeze
T_Field = TypeVar("T_Field")
class LAMField(Generic[T_Field]):
"""This class describe a Field in Schema"""
def __init__(self, name: str, v: Optional[T_Field], a: Any, i: LAMFieldInfo):
self._name: str = name
self._source: Optional[type] = None
self._default_value: Optional[T_Field] = v
self._value: Optional[T_Field] = v
self._annotations: Any = a
self._info: LAMFieldInfo = i
self._constraints: list[BaseConstraint[Any]] = i.constraints
def add_source(self, s: type) -> None:
"""Adds source Appliance to the Field"""
self._source = s
@property
def doc(self) -> str:
"""Returns Field's documentation"""
return self._info.doc
def add_constraint(self, c: BaseConstraint) -> None:
"""Adds constraint to the Field"""
self._constraints.append(c)
@property
def constraints(self) -> list[BaseConstraint]:
"""Returns Field's constraint"""
return self._info.constraints
@property
def default_value(self) -> Any:
"""Returns Field's default value (frozen)"""
return LAMdeepfreeze(self._default_value)
def update_value(self, v: Optional[T_Field] = None) -> None:
"""Updates Field's value"""
self._value = v
@property
def value(self) -> Any:
"""Returns Field's value (frozen)"""
return LAMdeepfreeze(self._value)
@property
def raw_value(self) -> Optional[T_Field]:
"""Returns Field's value"""
return self._value
@property
def annotations(self) -> Any:
"""Returns Field's annotation"""
return self._annotations

26
src/dabmodel/LAMFields/LAMFieldInfo.py Normal file
View File

@@ -0,0 +1,26 @@
from typing import Optional, Any
from .Constraint import BaseConstraint
class LAMFieldInfo:
"""This Class allows to describe a Field in Appliance class"""
def __init__(
self, *, doc: str = "", constraints: Optional[list[BaseConstraint]] = None
):
self._doc: str = doc
self._constraints: list[BaseConstraint]
if constraints is None:
self._constraints = []
else:
self._constraints = constraints
@property
def doc(self) -> str:
"""Returns Field's documentation"""
return self._doc
@property
def constraints(self) -> list[BaseConstraint[Any]]:
"""Returns Field's constraints"""
return self._constraints

16
src/dabmodel/__init__.py
View File

@@ -11,11 +11,16 @@ Main module __init__ file.
"""
from .__metadata__ import __version__, __Summuary__, __Name__
from .model import (
DABFieldInfo,
DABField,
BaseAppliance,
BaseFeature,
from .LAMFields.LAMField import LAMField
from .LAMFields.LAMFieldInfo import LAMFieldInfo
from .LAMFields.FrozenLAMField import FrozenLAMField
from .appliance import BaseAppliance
from .feature import BaseFeature
from .exception import (
DABModelException,
MultipleInheritanceForbidden,
BrokenInheritance,
@@ -28,7 +33,6 @@ from .model import (
IncompletelyAnnotatedField,
ImportForbidden,
FunctionForbidden,
FrozenDABField,
InvalidFeatureInheritance,
FeatureNotBound,
)

9
src/dabmodel/appliance.py Normal file
View File

@@ -0,0 +1,9 @@
from .element import BaseElement
from .meta.appliance import BaseMetaAppliance
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.
"""

7
src/dabmodel/element.py Normal file
View File

@@ -0,0 +1,7 @@
from .meta.base import BaseMeta
class BaseElement(metaclass=BaseMeta):
"""BaseElement class
Base class to apply metaclass and set common Fields.
"""

103
src/dabmodel/exception.py Normal file
View File

@@ -0,0 +1,103 @@
class DABModelException(Exception):
"""DABModelException Exception class
Base Exception for DABModelException class
"""
class FunctionForbidden(DABModelException): ...
class ExternalCodeForbidden(FunctionForbidden): ...
class ClosureForbidden(FunctionForbidden): ...
class ReservedFieldName(Exception):
"""DABModelException Exception class
Base Exception for DABModelException class
"""
class MultipleInheritanceForbidden(DABModelException):
"""MultipleInheritanceForbidden Exception class
Multiple inheritance is forbidden when using dabmodel
"""
class BrokenInheritance(DABModelException):
"""BrokenInheritance Exception class
inheritance chain is broken
"""
class ReadOnlyField(DABModelException):
"""ReadOnlyField Exception class
The used Field is ReadOnly
"""
class NewFieldForbidden(DABModelException):
"""NewFieldForbidden Exception class
Field creation is forbidden
"""
class InvalidFieldAnnotation(DABModelException):
"""InvalidFieldAnnotation Exception class
The field annotation is invalid
"""
class InvalidInitializerType(DABModelException):
"""InvalidInitializerType Exception class
The initializer is not a valid type
"""
class NotAnnotatedField(InvalidFieldAnnotation):
"""NotAnnotatedField Exception class
The Field is not Annotated
"""
class IncompletelyAnnotatedField(InvalidFieldAnnotation):
"""IncompletelyAnnotatedField Exception class
The field annotation is incomplete
"""
class ReadOnlyFieldAnnotation(DABModelException):
"""ReadOnlyFieldAnnotation Exception class
Field annotation connot be modified
"""
class InvalidFieldValue(DABModelException):
"""InvalidFieldValue Exception class
The Field value is invalid
"""
class NonExistingField(DABModelException):
"""NonExistingField Exception class
The given Field is non existing
"""
class ImportForbidden(DABModelException):
"""ImportForbidden Exception class
Imports are forbidden
"""
class InvalidFeatureInheritance(DABModelException):
"""InvalidFeatureInheritance Exception class
Features of same name in child appliance need to be from same type
"""
class FeatureNotBound(DABModelException):
"""FeatureNotBound Exception class
a Feature must be bound to the appliance (or parent)
"""

12
src/dabmodel/feature.py Normal file
View File

@@ -0,0 +1,12 @@
from .element import BaseElement
from .meta.feature import BaseMetaFeature
class BaseFeature(BaseElement, metaclass=BaseMetaFeature):
"""BaseFeature class
Base class for Appliance's Features.
Features are optional traits of an appliance.
"""
_BoundAppliance: "Optional[Type[BaseAppliance]]" = None
Enabled: bool = False

235
src/dabmodel/meta/appliance.py Normal file
View File

@@ -0,0 +1,235 @@
from typing import Any, Type
from copy import copy
from typeguard import check_type, CollectionCheckStrategy, TypeCheckError
from ..tools import LAMdeepfreeze
from ..LAMFields.LAMField import LAMField
from ..LAMFields.FrozenLAMField import FrozenLAMField
from .base import BaseMeta
from ..feature import BaseFeature
from ..exception import (
InvalidFieldValue,
InvalidFeatureInheritance,
FeatureNotBound,
)
class BaseMetaAppliance(BaseMeta):
"""BaseMetaAppliance class
Appliance specific metaclass code
"""
@classmethod
def check_class(
mcs: type["meta"],
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any], # pylint: disable=unused-argument
extensions: dict[str, Any],
) -> None:
"""
Appliance-specific pre-check: ensure the `features` slot exists in schema.
Copies the parent's `features` mapping when inheriting to keep it per-class.
"""
super().check_class(name, bases, namespace, extensions) # type: ignore[misc]
if "features" not in namespace["__DABSchema__"]:
namespace["__DABSchema__"]["features"] = {}
else:
namespace["__DABSchema__"]["features"] = copy(
namespace["__DABSchema__"]["features"]
)
@classmethod
def process_class_fields(
mcs: type["meta"],
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any],
extensions: dict[str, Any],
):
"""
Like meta.process_class_fields but also stages Feature declarations.
Initializes:
extensions["new_features"], extensions["modified_features"]
then defers to the base scanner for regular fields.
"""
extensions["new_features"] = {}
extensions["modified_features"] = {}
super().process_class_fields(name, bases, namespace, extensions) # type: ignore[misc]
@classmethod
def process_new_field(
mcs: type["meta"],
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any],
_fname: str,
_fvalue: Any,
extensions: dict[str, Any],
): # pylint: disable=unused-argument
"""
Intercept Feature declarations.
- If `_fname` already exists in parent's `features`, enforce same type;
stage into `modified_features`.
- Else, if `_fvalue` is a Feature *class*, stage into `new_features`.
- Otherwise, it is a regular field: delegate to meta.process_new_field.
"""
if _fname in namespace["__DABSchema__"]["features"].keys():
if not issubclass(_fvalue, namespace["__DABSchema__"]["features"][_fname]):
raise InvalidFeatureInheritance(
f"Feature {_fname} is not an instance of {bases[0]}.{_fname}"
)
extensions["modified_features"][_fname] = _fvalue
elif isinstance(_fvalue, type) and issubclass(_fvalue, BaseFeature):
extensions["new_features"][_fname] = _fvalue
else:
super().process_new_field(name, bases, namespace, _fname, _fvalue, extensions) # type: ignore[misc]
@classmethod
def commit_fields(
mcs: type["meta"],
cls,
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any], # pylint: disable=unused-argument
extensions: dict[str, Any],
):
"""
Commit regular fields (via meta) and then bind staged Feature classes.
For each new/modified feature:
- bind it to `cls` (sets the feature's `_BoundAppliance`),
- register it under `cls.__DABSchema__["features"]`.
"""
super().commit_fields(cls, name, bases, namespace, extensions) # type: ignore[misc]
for _ftname, _ftvalue in extensions["modified_features"].items():
_ftvalue._BoundAppliance = cls # pylint: disable=protected-access
cls.__DABSchema__["features"][_ftname] = _ftvalue
for _ftname, _ftvalue in extensions["new_features"].items():
_ftvalue._BoundAppliance = cls # pylint: disable=protected-access
cls.__DABSchema__["features"][_ftname] = _ftvalue
def finalize_instance(cls: Type, obj, extensions: dict[str, Any]):
"""
Instantiate and attach all features declared (or overridden) in the instance schema.
Handles:
- Declared features (plain class)
- Subclass replacements
- Dict overrides (class + patch dict)
"""
for fname, fdef in obj.__DABSchema__.get("features", {}).items():
# Case 1: plain class or subclass
if isinstance(fdef, type) and issubclass(fdef, BaseFeature):
inst = fdef()
object.__setattr__(obj, fname, inst)
# Case 2: (class, dict) → dict overrides
elif isinstance(fdef, tuple) and len(fdef) == 2:
feat_cls, overrides = fdef
inst = feat_cls()
for field_name, new_val in overrides.items():
if field_name not in feat_cls.__DABSchema__:
raise InvalidFieldValue(
f"Feature '{fname}' has no field '{field_name}'"
)
field = feat_cls.__DABSchema__[field_name]
try:
check_type(
new_val,
field.annotations,
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
)
except TypeCheckError as exp:
raise InvalidFieldValue(
f"Invalid value for {fname}.{field_name}: "
f"expected {field.annotations}, got {new_val!r}"
) from exp
object.__setattr__(inst, field_name, LAMdeepfreeze(new_val))
inst.__DABSchema__[field_name] = FrozenLAMField(
LAMField(field_name, new_val, field.annotations, field._info)
)
object.__setattr__(obj, fname, inst)
else:
raise InvalidFieldValue(
f"Invalid feature definition stored for '{fname}': {fdef!r}"
)
def apply_overrides(cls, obj, extensions, *args, **kwargs):
"""
Support for runtime field and feature overrides.
Fields:
MyApp(name="foo")
Features:
MyApp(F1=MyF1) # inheritance / replacement
MyApp(F1={"val": 42, ...}) # dict override of existing feature
"""
# --- field overrides (unchanged) ---
for k, v in list(kwargs.items()):
if k in cls.__DABSchema__: # regular field
field = cls.__DABSchema__[k]
try:
check_type(
v,
field.annotations,
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
)
except TypeCheckError as exp:
raise InvalidFieldValue(
f"Invalid value for field '{k}': expected {field.annotations}, got {v!r}"
) from exp
object.__setattr__(obj, k, LAMdeepfreeze(v))
obj.__DABSchema__[k] = FrozenLAMField(
LAMField(k, v, field.annotations, field._info)
)
kwargs.pop(k)
# --- feature overrides ---
for k, v in list(kwargs.items()):
if k in cls.__DABSchema__.get("features", {}):
base_feat_cls = cls.__DABSchema__["features"][k]
# Case 1: subclass replacement (inheritance)
if isinstance(v, type) and issubclass(v, base_feat_cls):
bound = getattr(v, "_BoundAppliance", None)
if bound is None or not issubclass(cls, bound):
raise FeatureNotBound(
f"Feature {v.__name__} is not bound to {cls.__name__}"
)
# record subclass into instance schema
obj.__DABSchema__["features"][k] = v
kwargs.pop(k)
# Case 2: dict override
elif isinstance(v, dict):
# store (class, override_dict) for finalize_instance
obj.__DABSchema__["features"][k] = (base_feat_cls, v)
kwargs.pop(k)
else:
raise InvalidFieldValue(
f"Feature override for '{k}' must be a Feature subclass or dict, got {type(v)}"
)
# --- new features not declared at class level ---
for k, v in list(kwargs.items()):
if isinstance(v, type) and issubclass(v, BaseFeature):
bound = getattr(v, "_BoundAppliance", None)
if bound is None or not issubclass(cls, bound):
raise FeatureNotBound(
f"Feature {v.__name__} is not bound to {cls.__name__}"
)
obj.__DABSchema__["features"][k] = v
kwargs.pop(k)
if kwargs:
unknown = ", ".join(sorted(kwargs.keys()))
raise InvalidFieldValue(f"Unknown parameters: {unknown}")

568
src/dabmodel/model.py → src/dabmodel/meta/base.py
View File

@@ -1,13 +1,6 @@
"""dabmodel model module
This module implements DAB model classes.
This module contains metaclass and bases classes used to create models.
BaseAppliance can be used to create a new Appliance Data.
BaseFeature can be used to create new Appliance's Features."""
from typing import (
Optional,
TypeVar,
Generic,
Union,
get_origin,
get_args,
@@ -21,16 +14,35 @@ from typing import (
Callable,
Type,
)
from types import UnionType, FunctionType, SimpleNamespace
from copy import deepcopy, copy
# from pprint import pprint
import math
import inspect, ast, textwrap
from frozendict import deepfreeze
from typeguard import check_type, TypeCheckError, CollectionCheckStrategy
from ..LAMFields.LAMField import LAMField
from ..LAMFields.LAMFieldInfo import LAMFieldInfo
from ..LAMFields.FrozenLAMField import FrozenLAMField
from ..exception import (
MultipleInheritanceForbidden,
BrokenInheritance,
ReadOnlyField,
NewFieldForbidden,
NotAnnotatedField,
ReadOnlyFieldAnnotation,
InvalidFieldValue,
InvalidFieldAnnotation,
IncompletelyAnnotatedField,
ImportForbidden,
FunctionForbidden,
NonExistingField,
InvalidInitializerType,
)
ALLOWED_ANNOTATIONS: dict[str, Any] = {
"Union": Union,
"Optional": Optional,
@@ -54,7 +66,6 @@ ALLOWED_ANNOTATIONS: dict[str, Any] = {
"frozenset": frozenset,
"tuple": tuple,
}
ALLOWED_MODEL_FIELDS_TYPES: tuple[type[Any], ...] = (
str,
int,
@@ -65,111 +76,6 @@ ALLOWED_MODEL_FIELDS_TYPES: tuple[type[Any], ...] = (
)
class DABModelException(Exception):
"""DABModelException Exception class
Base Exception for DABModelException class
"""
class FunctionForbidden(DABModelException): ...
class ExternalCodeForbidden(FunctionForbidden): ...
class ClosureForbidden(FunctionForbidden): ...
class ReservedFieldName(Exception):
"""DABModelException Exception class
Base Exception for DABModelException class
"""
class MultipleInheritanceForbidden(DABModelException):
"""MultipleInheritanceForbidden Exception class
Multiple inheritance is forbidden when using dabmodel
"""
class BrokenInheritance(DABModelException):
"""BrokenInheritance Exception class
inheritance chain is broken
"""
class ReadOnlyField(DABModelException):
"""ReadOnlyField Exception class
The used Field is ReadOnly
"""
class NewFieldForbidden(DABModelException):
"""NewFieldForbidden Exception class
Field creation is forbidden
"""
class InvalidFieldAnnotation(DABModelException):
"""InvalidFieldAnnotation Exception class
The field annotation is invalid
"""
class InvalidInitializerType(DABModelException):
"""InvalidInitializerType Exception class
The initializer is not a valid type
"""
class NotAnnotatedField(InvalidFieldAnnotation):
"""NotAnnotatedField Exception class
The Field is not Annotated
"""
class IncompletelyAnnotatedField(InvalidFieldAnnotation):
"""IncompletelyAnnotatedField Exception class
The field annotation is incomplete
"""
class ReadOnlyFieldAnnotation(DABModelException):
"""ReadOnlyFieldAnnotation Exception class
Field annotation connot be modified
"""
class InvalidFieldValue(DABModelException):
"""InvalidFieldValue Exception class
The Field value is invalid
"""
class NonExistingField(DABModelException):
"""NonExistingField Exception class
The given Field is non existing
"""
class ImportForbidden(DABModelException):
"""ImportForbidden Exception class
Imports are forbidden
"""
class InvalidFeatureInheritance(DABModelException):
"""InvalidFeatureInheritance Exception class
Features of same name in child appliance need to be from same type
"""
class FeatureNotBound(DABModelException):
"""FeatureNotBound Exception class
a Feature must be bound to the appliance (or parent)
"""
ALLOWED_HELPERS_MATH = SimpleNamespace(
sqrt=math.sqrt,
floor=math.floor,
@@ -358,149 +264,6 @@ def _check_annotation_definition( # pylint: disable=too-complex,too-many-return
return False
T_Field = TypeVar("T_Field")
class BaseConstraint(Generic[T_Field]):
"""BaseConstraint class
Base class for Field's constraints
"""
_bound_type: type
def __init__(self): ...
def check(self, value: T_Field) -> bool:
"""Check if a Constraint is completed"""
return True
def _deepfreeze(value):
"""recursive freeze helper function"""
if isinstance(value, dict):
return deepfreeze(value)
if isinstance(value, set):
return frozenset(_deepfreeze(v) for v in value)
if isinstance(value, list):
return tuple(_deepfreeze(v) for v in value)
if isinstance(value, tuple):
return tuple(_deepfreeze(v) for v in value)
return value
class DABFieldInfo:
"""This Class allows to describe a Field in Appliance class"""
def __init__(
self, *, doc: str = "", constraints: Optional[list[BaseConstraint]] = None
):
self._doc: str = doc
self._constraints: list[BaseConstraint]
if constraints is None:
self._constraints = []
else:
self._constraints = constraints
@property
def doc(self) -> str:
"""Returns Field's documentation"""
return self._doc
@property
def constraints(self) -> list[BaseConstraint[Any]]:
"""Returns Field's constraints"""
return self._constraints
class DABField(Generic[T_Field]):
"""This class describe a Field in Schema"""
def __init__(self, name: str, v: Optional[T_Field], a: Any, i: DABFieldInfo):
self._name: str = name
self._source: Optional[type] = None
self._default_value: Optional[T_Field] = v
self._value: Optional[T_Field] = v
self._annotations: Any = a
self._info: DABFieldInfo = i
self._constraints: List[BaseConstraint[Any]] = i.constraints
def add_source(self, s: type) -> None:
"""Adds source Appliance to the Field"""
self._source = s
@property
def doc(self) -> str:
"""Returns Field's documentation"""
return self._info.doc
def add_constraint(self, c: BaseConstraint) -> None:
"""Adds constraint to the Field"""
self._constraints.append(c)
@property
def constraints(self) -> list[BaseConstraint]:
"""Returns Field's constraint"""
return self._info.constraints
@property
def default_value(self) -> Any:
"""Returns Field's default value (frozen)"""
return _deepfreeze(self._default_value)
def update_value(self, v: Optional[T_Field] = None) -> None:
"""Updates Field's value"""
self._value = v
@property
def value(self) -> Any:
"""Returns Field's value (frosen)"""
return _deepfreeze(self._value)
@property
def raw_value(self) -> Optional[T_Field]:
"""Returns Field's value"""
return self._value
@property
def annotations(self) -> Any:
"""Returns Field's annotation"""
return self._annotations
class FrozenDABField(Generic[T_Field]):
"""FrozenDABField class
a read-only proxy of a Field
"""
def __init__(self, inner_field: DABField):
self._inner_field = inner_field
@property
def doc(self) -> str:
"""Returns Field's documentation (frozen)"""
return _deepfreeze(self._inner_field.doc)
@property
def constraints(self) -> tuple[BaseConstraint]:
"""Returns Field's constraint (frozen)"""
return _deepfreeze(self._inner_field.constraints)
@property
def default_value(self) -> Any:
"""Returns Field's default value (frozen)"""
return self._inner_field.default_value
@property
def value(self) -> Any:
"""Returns Field's value (frosen)"""
return self._inner_field.value
@property
def annotations(self) -> Any:
"""Returns Field's annotation (frozen)"""
return _deepfreeze(self._inner_field.annotations)
class ModelSpecView:
"""ModelSpecView class
A class that will act as fake BaseElement proxy to allow setting values"""
@@ -572,7 +335,7 @@ class BaseMeta(type):
"""metaclass to use to build BaseElement"""
modified_fields: Dict[str, Any] = {}
new_fields: Dict[str, DABField[Any]] = {}
new_fields: Dict[str, LAMField[Any]] = {}
initializer: Optional[Callable[..., Any]] = None
__DABSchema__: dict[str, Any] = {}
@@ -724,8 +487,8 @@ class BaseMeta(type):
Handle a *new* field declared on the class.
Resolves string annotations against a whitelist, validates `Annotated[...]`
payloads (allowing only DABFieldInfo), checks the default value type,
and stages the field as a `DABField` in `mcs.new_fields`.
payloads (allowing only LAMFieldInfo), checks the default value type,
and stages the field as a `LAMField` in `mcs.new_fields`.
"""
# print(f"New field: {_fname}")
@@ -749,7 +512,7 @@ class BaseMeta(type):
f"Field <{_fname}> has not an allowed or valid annotation."
)
_finfo: DABFieldInfo = DABFieldInfo()
_finfo: LAMFieldInfo = LAMFieldInfo()
origin = get_origin(namespace["__annotations__"][_fname])
tname = (
getattr(origin, "__name__", "")
@@ -763,9 +526,9 @@ class BaseMeta(type):
raise InvalidFieldAnnotation(
f"Field <{_fname}> had invalid Annotated value."
)
if len(args) == 2 and not isinstance(args[1], DABFieldInfo):
if len(args) == 2 and not isinstance(args[1], LAMFieldInfo):
raise InvalidFieldAnnotation(
"Only DABFieldInfo object is allowed as Annotated data."
"Only LAMFieldInfo object is allowed as Annotated data."
)
_finfo = args[1]
@@ -781,7 +544,7 @@ 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(
mcs.new_fields[_fname] = LAMField(
_fname, _fvalue, namespace["__annotations__"][_fname], _finfo
)
@@ -807,7 +570,7 @@ class BaseMeta(type):
init_fieldvalues = {}
init_fieldtypes = {}
for _fname, _fvalue in cls.__DABSchema__.items():
if isinstance(_fvalue, DABField):
if isinstance(_fvalue, LAMField):
init_fieldvalues[_fname] = deepcopy(_fvalue.raw_value)
init_fieldtypes[_fname] = _fvalue.annotations
fakecls = ModelSpecView(
@@ -871,8 +634,8 @@ class BaseMeta(type):
"""
Commit staged fields into the class schema (`__DABSchema__`).
- For modified fields: copy the parent's DABField, update its value.
- For new fields: set the freshly built DABField and record its source.
- For modified fields: copy the parent's LAMField, update its value.
- For new fields: set the freshly built LAMField and record its source.
"""
for _fname, _fvalue in mcs.modified_fields.items():
cls.__DABSchema__[_fname] = deepcopy(bases[0].__DABSchema__[_fname])
@@ -907,24 +670,24 @@ class BaseMeta(type):
"""
Populate the new instance with field values from the class schema.
Copies each DABField.value to an instance attribute (deep-frozen view).
Copies each LAMField.value to an instance attribute (deep-frozen view).
"""
for _fname, _fvalue in cls.__DABSchema__.items():
if isinstance(_fvalue, DABField):
if isinstance(_fvalue, LAMField):
object.__setattr__(obj, _fname, _fvalue.value)
def freeze_instance_schema(
cls: Type, obj: Any, extensions: dict[str, Any], *args: Any, **kw: Any
):
"""
Freeze the instance's schema by wrapping DABFields into FrozenDABField.
Freeze the instance's schema by wrapping DABFields into FrozenLAMField.
Creates a per-instance `__DABSchema__` dict where each field is read-only.
"""
inst_schema = copy(obj.__DABSchema__)
for _fname, _fvalue in cls.__DABSchema__.items():
if isinstance(_fvalue, DABField):
inst_schema[_fname] = FrozenDABField(_fvalue)
if isinstance(_fvalue, LAMField):
inst_schema[_fname] = FrozenLAMField(_fvalue)
if "features" in inst_schema:
inst_schema["features"] = dict(inst_schema["features"])
@@ -984,262 +747,3 @@ class BaseMeta(type):
return orig_setattr(_self, key, value)
setattr(cls, "__setattr__", guarded_setattr)
class BaseElement(metaclass=BaseMeta):
"""BaseElement class
Base class to apply metaclass and set common Fields.
"""
class BaseMetaFeature(BaseMeta):
"""BaseMetaFeature class
Feature specific metaclass code
"""
def __call__(cls: Type, *args: Any, **kw: Any): # intentionally untyped
"""BaseFeature new instance"""
if cls._BoundAppliance is None:
raise FeatureNotBound()
obj = super().__call__(*args, **kw)
return obj
class BaseFeature(BaseElement, metaclass=BaseMetaFeature):
"""BaseFeature class
Base class for Appliance's Features.
Features are optional traits of an appliance.
"""
_BoundAppliance: "Optional[Type[BaseAppliance]]" = None
Enabled: bool = False
class BaseMetaAppliance(BaseMeta):
"""BaseMetaAppliance class
Appliance specific metaclass code
"""
@classmethod
def check_class(
mcs: type["BaseMeta"],
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any], # pylint: disable=unused-argument
extensions: dict[str, Any],
) -> None:
"""
Appliance-specific pre-check: ensure the `features` slot exists in schema.
Copies the parent's `features` mapping when inheriting to keep it per-class.
"""
super().check_class(name, bases, namespace, extensions) # type: ignore[misc]
if "features" not in namespace["__DABSchema__"]:
namespace["__DABSchema__"]["features"] = {}
else:
namespace["__DABSchema__"]["features"] = copy(
namespace["__DABSchema__"]["features"]
)
@classmethod
def process_class_fields(
mcs: type["BaseMeta"],
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any],
extensions: dict[str, Any],
):
"""
Like BaseMeta.process_class_fields but also stages Feature declarations.
Initializes:
extensions["new_features"], extensions["modified_features"]
then defers to the base scanner for regular fields.
"""
extensions["new_features"] = {}
extensions["modified_features"] = {}
super().process_class_fields(name, bases, namespace, extensions) # type: ignore[misc]
@classmethod
def process_new_field(
mcs: type["BaseMeta"],
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any],
_fname: str,
_fvalue: Any,
extensions: dict[str, Any],
): # pylint: disable=unused-argument
"""
Intercept Feature declarations.
- If `_fname` already exists in parent's `features`, enforce same type;
stage into `modified_features`.
- Else, if `_fvalue` is a Feature *class*, stage into `new_features`.
- Otherwise, it is a regular field: delegate to BaseMeta.process_new_field.
"""
if _fname in namespace["__DABSchema__"]["features"].keys():
if not issubclass(_fvalue, namespace["__DABSchema__"]["features"][_fname]):
raise InvalidFeatureInheritance(
f"Feature {_fname} is not an instance of {bases[0]}.{_fname}"
)
extensions["modified_features"][_fname] = _fvalue
elif isinstance(_fvalue, type) and issubclass(_fvalue, BaseFeature):
extensions["new_features"][_fname] = _fvalue
else:
super().process_new_field(name, bases, namespace, _fname, _fvalue, extensions) # type: ignore[misc]
@classmethod
def commit_fields(
mcs: type["BaseMeta"],
cls,
name: str,
bases: tuple[type[Any], ...],
namespace: dict[str, Any], # pylint: disable=unused-argument
extensions: dict[str, Any],
):
"""
Commit regular fields (via BaseMeta) and then bind staged Feature classes.
For each new/modified feature:
- bind it to `cls` (sets the feature's `_BoundAppliance`),
- register it under `cls.__DABSchema__["features"]`.
"""
super().commit_fields(cls, name, bases, namespace, extensions) # type: ignore[misc]
for _ftname, _ftvalue in extensions["modified_features"].items():
_ftvalue._BoundAppliance = cls # pylint: disable=protected-access
cls.__DABSchema__["features"][_ftname] = _ftvalue
for _ftname, _ftvalue in extensions["new_features"].items():
_ftvalue._BoundAppliance = cls # pylint: disable=protected-access
cls.__DABSchema__["features"][_ftname] = _ftvalue
def finalize_instance(cls: Type, obj, extensions: dict[str, Any]):
"""
Instantiate and attach all features declared (or overridden) in the instance schema.
Handles:
- Declared features (plain class)
- Subclass replacements
- Dict overrides (class + patch dict)
"""
for fname, fdef in obj.__DABSchema__.get("features", {}).items():
# Case 1: plain class or subclass
if isinstance(fdef, type) and issubclass(fdef, BaseFeature):
inst = fdef()
object.__setattr__(obj, fname, inst)
# Case 2: (class, dict) → dict overrides
elif isinstance(fdef, tuple) and len(fdef) == 2:
feat_cls, overrides = fdef
inst = feat_cls()
for field_name, new_val in overrides.items():
if field_name not in feat_cls.__DABSchema__:
raise InvalidFieldValue(
f"Feature '{fname}' has no field '{field_name}'"
)
field = feat_cls.__DABSchema__[field_name]
try:
check_type(
new_val,
field.annotations,
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
)
except TypeCheckError as exp:
raise InvalidFieldValue(
f"Invalid value for {fname}.{field_name}: "
f"expected {field.annotations}, got {new_val!r}"
) from exp
object.__setattr__(inst, field_name, _deepfreeze(new_val))
inst.__DABSchema__[field_name] = FrozenDABField(
DABField(field_name, new_val, field.annotations, field._info)
)
object.__setattr__(obj, fname, inst)
else:
raise InvalidFieldValue(
f"Invalid feature definition stored for '{fname}': {fdef!r}"
)
def apply_overrides(cls, obj, extensions, *args, **kwargs):
"""
Support for runtime field and feature overrides.
Fields:
MyApp(name="foo")
Features:
MyApp(F1=MyF1) # inheritance / replacement
MyApp(F1={"val": 42, ...}) # dict override of existing feature
"""
# --- field overrides (unchanged) ---
for k, v in list(kwargs.items()):
if k in cls.__DABSchema__: # regular field
field = cls.__DABSchema__[k]
try:
check_type(
v,
field.annotations,
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
)
except TypeCheckError as exp:
raise InvalidFieldValue(
f"Invalid value for field '{k}': expected {field.annotations}, got {v!r}"
) from exp
object.__setattr__(obj, k, _deepfreeze(v))
obj.__DABSchema__[k] = FrozenDABField(
DABField(k, v, field.annotations, field._info)
)
kwargs.pop(k)
# --- feature overrides ---
for k, v in list(kwargs.items()):
if k in cls.__DABSchema__.get("features", {}):
base_feat_cls = cls.__DABSchema__["features"][k]
# Case 1: subclass replacement (inheritance)
if isinstance(v, type) and issubclass(v, base_feat_cls):
bound = getattr(v, "_BoundAppliance", None)
if bound is None or not issubclass(cls, bound):
raise FeatureNotBound(
f"Feature {v.__name__} is not bound to {cls.__name__}"
)
# record subclass into instance schema
obj.__DABSchema__["features"][k] = v
kwargs.pop(k)
# Case 2: dict override
elif isinstance(v, dict):
# store (class, override_dict) for finalize_instance
obj.__DABSchema__["features"][k] = (base_feat_cls, v)
kwargs.pop(k)
else:
raise InvalidFieldValue(
f"Feature override for '{k}' must be a Feature subclass or dict, got {type(v)}"
)
# --- new features not declared at class level ---
for k, v in list(kwargs.items()):
if isinstance(v, type) and issubclass(v, BaseFeature):
bound = getattr(v, "_BoundAppliance", None)
if bound is None or not issubclass(cls, bound):
raise FeatureNotBound(
f"Feature {v.__name__} is not bound to {cls.__name__}"
)
obj.__DABSchema__["features"][k] = v
kwargs.pop(k)
if kwargs:
unknown = ", ".join(sorted(kwargs.keys()))
raise InvalidFieldValue(f"Unknown parameters: {unknown}")
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.
"""

19
src/dabmodel/meta/feature.py Normal file
View File

@@ -0,0 +1,19 @@
from typing import Type, Any
from .base import BaseMeta
from ..exception import FeatureNotBound
class BaseMetaFeature(BaseMeta):
"""BaseMetaFeature class
Feature specific metaclass code
"""
def __call__(cls: Type, *args: Any, **kw: Any): # intentionally untyped
"""BaseFeature new instance"""
if cls._BoundAppliance is None:
raise FeatureNotBound()
obj = super().__call__(*args, **kw)
return obj

17
src/dabmodel/tools.py
View File

@@ -4,8 +4,10 @@ from uuid import UUID
from datetime import datetime
import json
from frozendict import deepfreeze
class DABJSONEncoder(json.JSONEncoder):
class LAMJSONEncoder(json.JSONEncoder):
"""allows to JSON encode non supported data type"""
def default(self, o):
@@ -15,3 +17,16 @@ class DABJSONEncoder(json.JSONEncoder):
if isinstance(o, datetime):
return str(o)
return json.JSONEncoder.default(self, o)
def LAMdeepfreeze(value):
"""recursive freeze helper function"""
if isinstance(value, dict):
return deepfreeze(value)
if isinstance(value, set):
return frozenset(LAMdeepfreeze(v) for v in value)
if isinstance(value, list):
return tuple(LAMdeepfreeze(v) for v in value)
if isinstance(value, tuple):
return tuple(LAMdeepfreeze(v) for v in value)
return value

93
test/test_model.py
View File

@@ -21,13 +21,10 @@ from typing import (
Tuple,
Set,
FrozenSet,
TypeVar,
Generic,
Any,
Annotated,
)
from pprint import pprint
from frozendict import frozendict
import math
print(__name__)
@@ -256,7 +253,7 @@ class MainTests(unittest.TestCase):
self.immutable_vars__test_field(app2, "StrVar5", "default value", "123")
self.immutable_vars__test_field(app2, "StrVar6", None, "123")
@unittest.skip
# @unittest.skip
def test_containers__set(self):
"""Testing first appliance level, and Field types (Set)"""
@@ -297,23 +294,23 @@ class MainTests(unittest.TestCase):
with self.assertRaises(dm.InvalidFieldValue):
class _(dm.BaseAppliance):
_: Set[int] = {"a"}
class A(dm.BaseAppliance):
a: Set[int] = {"a"}
with self.assertRaises(dm.InvalidFieldValue):
class _(dm.BaseAppliance):
_: "Set[int]" = {"a"}
class B(dm.BaseAppliance):
b: "Set[int]" = {"a"}
with self.assertRaises(dm.IncompletelyAnnotatedField):
class _(dm.BaseAppliance):
_: Set = {1, 2}
class C(dm.BaseAppliance):
c: Set = {1, 2}
with self.assertRaises(dm.IncompletelyAnnotatedField):
class _(dm.BaseAppliance):
_: "Set" = {1, 2}
class D(dm.BaseAppliance):
d: "Set" = {1, 2}
# Hacky part !
# set() are randomly ordered, and typeguard can be configured to only check 1st element of containers.
@@ -339,7 +336,7 @@ class MainTests(unittest.TestCase):
res = subprocess.run([sys.executable, "-c", code], env=env)
self.assertEqual(res.returncode, 2)
@unittest.skip
# @unittest.skip
def test_containers__frozenset(self):
"""Testing first appliance level, and Field types (FrozenSet)"""
@@ -373,25 +370,27 @@ class MainTests(unittest.TestCase):
with self.assertRaises(AttributeError):
app1.testVar.add(3)
# /!\ THIS SHOULD RAISE AN EXCEPTION BUT IT DOESNT :-/
with self.assertRaises(dm.InvalidFieldValue):
print("youhou")
class _(dm.BaseAppliance):
_: FrozenSet[int] = {"a"}
class A(dm.BaseAppliance):
a: FrozenSet[int] = {"a"}
with self.assertRaises(dm.InvalidFieldValue):
class _(dm.BaseAppliance):
_: "FrozenSet[int]" = {"a"}
class B(dm.BaseAppliance):
b: "FrozenSet[int]" = {"a"}
with self.assertRaises(dm.IncompletelyAnnotatedField):
class _(dm.BaseAppliance):
_: FrozenSet = {1, 2}
class C(dm.BaseAppliance):
c: FrozenSet = {1, 2}
with self.assertRaises(dm.IncompletelyAnnotatedField):
class _(dm.BaseAppliance):
_: "FrozenSet" = {1, 2}
class D(dm.BaseAppliance):
d: "FrozenSet" = {1, 2}
# Hacky part !
# set() are randomly ordered, and typeguard can be configured to only check 1st element of containers.
@@ -461,22 +460,22 @@ class MainTests(unittest.TestCase):
with self.assertRaises(dm.InvalidFieldValue):
class _(dm.BaseAppliance):
class E(dm.BaseAppliance):
test: List[int] = ["a"]
with self.assertRaises(dm.InvalidFieldValue):
class _(dm.BaseAppliance):
class F(dm.BaseAppliance):
test: "List[int]" = ["a"]
with self.assertRaises(dm.IncompletelyAnnotatedField):
class _(dm.BaseAppliance):
class G(dm.BaseAppliance):
test: List = [1, 2]
with self.assertRaises(dm.IncompletelyAnnotatedField):
class _(dm.BaseAppliance):
class H(dm.BaseAppliance):
test: "List" = [1, 2]
def test_containers__dict(self):
@@ -790,20 +789,20 @@ class MainTests(unittest.TestCase):
# class can be created
class Appliance1(dm.BaseAppliance):
StrVar: Annotated[str, dm.DABFieldInfo(doc="foo1")] = "default value"
StrVar2: Annotated[str, dm.DABFieldInfo(doc="foo2")] = "default value2"
VarInt: Annotated[int, dm.DABFieldInfo(doc="foo3")] = 12
VarInt2: Annotated[int, dm.DABFieldInfo(doc="foo4")] = 21
VarFloat: Annotated[float, dm.DABFieldInfo(doc="foo5")] = 12.1
VarFloat2: Annotated[float, dm.DABFieldInfo(doc="foo6")] = 21.2
VarComplex: Annotated[complex, dm.DABFieldInfo(doc="foo7")] = complex(3, 5)
VarComplex2: Annotated[complex, dm.DABFieldInfo(doc="foo8")] = complex(8, 6)
VarBool: Annotated[bool, dm.DABFieldInfo(doc="foo9")] = True
VarBool2: Annotated[bool, dm.DABFieldInfo(doc="foo10")] = False
VarBytes: Annotated[bytes, dm.DABFieldInfo(doc="foo11")] = bytes.fromhex(
StrVar: Annotated[str, dm.LAMFieldInfo(doc="foo1")] = "default value"
StrVar2: Annotated[str, dm.LAMFieldInfo(doc="foo2")] = "default value2"
VarInt: Annotated[int, dm.LAMFieldInfo(doc="foo3")] = 12
VarInt2: Annotated[int, dm.LAMFieldInfo(doc="foo4")] = 21
VarFloat: Annotated[float, dm.LAMFieldInfo(doc="foo5")] = 12.1
VarFloat2: Annotated[float, dm.LAMFieldInfo(doc="foo6")] = 21.2
VarComplex: Annotated[complex, dm.LAMFieldInfo(doc="foo7")] = complex(3, 5)
VarComplex2: Annotated[complex, dm.LAMFieldInfo(doc="foo8")] = complex(8, 6)
VarBool: Annotated[bool, dm.LAMFieldInfo(doc="foo9")] = True
VarBool2: Annotated[bool, dm.LAMFieldInfo(doc="foo10")] = False
VarBytes: Annotated[bytes, dm.LAMFieldInfo(doc="foo11")] = bytes.fromhex(
"2Ef0 F1f2 "
)
VarBytes2: Annotated[bytes, dm.DABFieldInfo(doc="foo12")] = bytes.fromhex(
VarBytes2: Annotated[bytes, dm.LAMFieldInfo(doc="foo12")] = bytes.fromhex(
"2ff0 F7f2 "
)
@@ -1639,19 +1638,19 @@ class MainTests(unittest.TestCase):
app1 = Appliance1()
self.assertIsInstance(Appliance1.__DABSchema__["VarStrOuter"], dm.DABField)
self.assertIsInstance(app1.__DABSchema__["VarStrOuter"], dm.FrozenDABField)
self.assertIsInstance(Appliance1.__DABSchema__["VarStrOuter"], dm.LAMField)
self.assertIsInstance(app1.__DABSchema__["VarStrOuter"], dm.FrozenLAMField)
self.assertIn("Feature1", app1.__DABSchema__["features"])
self.assertIn(
"VarStrInner", app1.__DABSchema__["features"]["Feature1"].__DABSchema__
)
self.assertIsInstance(
app1.__DABSchema__["features"]["Feature1"].__DABSchema__["VarStrInner"],
dm.DABField,
dm.LAMField,
)
self.assertTrue(hasattr(app1, "Feature1"))
self.assertIsInstance(
app1.Feature1.__DABSchema__["VarStrInner"], dm.FrozenDABField
app1.Feature1.__DABSchema__["VarStrInner"], dm.FrozenLAMField
)
self.assertTrue(hasattr(app1.Feature1, "VarStrInner"))
@@ -1691,19 +1690,19 @@ class MainTests(unittest.TestCase):
app2 = Appliance2()
app3 = Appliance3()
self.assertIsInstance(Appliance1.__DABSchema__["VarStrOuter"], dm.DABField)
self.assertIsInstance(app1.__DABSchema__["VarStrOuter"], dm.FrozenDABField)
self.assertIsInstance(Appliance1.__DABSchema__["VarStrOuter"], dm.LAMField)
self.assertIsInstance(app1.__DABSchema__["VarStrOuter"], dm.FrozenLAMField)
self.assertIn("Feature1", app1.__DABSchema__["features"])
self.assertIn(
"VarStrInner", app1.__DABSchema__["features"]["Feature1"].__DABSchema__
)
self.assertIsInstance(
app1.__DABSchema__["features"]["Feature1"].__DABSchema__["VarStrInner"],
dm.DABField,
dm.LAMField,
)
self.assertTrue(hasattr(app1, "Feature1"))
self.assertIsInstance(
app1.Feature1.__DABSchema__["VarStrInner"], dm.FrozenDABField
app1.Feature1.__DABSchema__["VarStrInner"], dm.FrozenLAMField
)
self.assertTrue(hasattr(app1.Feature1, "VarStrInner"))
self.assertEqual(app1.VarStrOuter, "testvalue APPLIANCE1")
@@ -2557,7 +2556,7 @@ class MainTests(unittest.TestCase):
class App(dm.BaseAppliance):
class F1(dm.BaseFeature):
a: Annotated[int, dm.DABFieldInfo(doc="field a")] = 1
a: Annotated[int, dm.LAMFieldInfo(doc="field a")] = 1
# ✅ Subclass override must inherit from parent F1
class F1Ex(App.F1):