Compare commits
28 Commits
0.0.1
...
0.0.1.post
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3e0defc574 | ||
|
|
f6e581381d | ||
|
|
981c5201a9 | ||
|
|
ab11052c8f | ||
|
|
4f5dade949 | ||
|
|
cce260bc5e | ||
|
|
915a4332ee | ||
|
|
4dca3eb9d1 | ||
|
|
e11c541139 | ||
|
|
637b50b325 | ||
|
|
f45c9cc8f3 | ||
|
|
95b0c298ce | ||
|
|
04a4cf7b36 | ||
|
|
f42a839cff | ||
|
|
7f3a4ef545 | ||
|
|
608c8a1010 | ||
|
|
210781f086 | ||
|
|
df966ccac4 | ||
|
|
29827b51bc | ||
|
|
7440731135 | ||
|
|
87682c2c9c | ||
|
|
0eef35e36f | ||
|
|
4c17436cea | ||
|
|
a90ab4885b | ||
|
|
192dcc74f8 | ||
|
|
171a2f1617 | ||
|
|
001ffbbbf1 | ||
|
|
8ab6c8e179 |
2
.project
2
.project
@@ -1,6 +1,6 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<projectDescription>
|
||||
<name>{{project_name}}</name>
|
||||
<name>dabmodel</name>
|
||||
<comment></comment>
|
||||
<projects>
|
||||
</projects>
|
||||
|
||||
@@ -1,13 +1,20 @@
|
||||
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
|
||||
<?eclipse-pydev version="1.0"?><pydev_project>
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_INTERPRETER">Default</pydev_property>
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_VERSION">python interpreter</pydev_property>
|
||||
|
||||
<pydev_pathproperty name="org.python.pydev.PROJECT_SOURCE_PATH">
|
||||
<path>/${PROJECT_DIR_NAME}/src</path>
|
||||
<path>/${PROJECT_DIR_NAME}</path>
|
||||
</pydev_pathproperty>
|
||||
|
||||
</pydev_project>
|
||||
<?xml version="1.0" encoding="UTF-8" standalone="no"?>
|
||||
<?eclipse-pydev version="1.0"?><pydev_project>
|
||||
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_INTERPRETER">Python311</pydev_property>
|
||||
|
||||
|
||||
<pydev_property name="org.python.pydev.PYTHON_PROJECT_VERSION">python interpreter</pydev_property>
|
||||
|
||||
|
||||
<pydev_pathproperty name="org.python.pydev.PROJECT_SOURCE_PATH">
|
||||
|
||||
<path>/${PROJECT_DIR_NAME}/src</path>
|
||||
|
||||
<path>/${PROJECT_DIR_NAME}</path>
|
||||
|
||||
</pydev_pathproperty>
|
||||
|
||||
|
||||
</pydev_project>
|
||||
|
||||
@@ -34,7 +34,9 @@ classifiers = [
|
||||
]
|
||||
dependencies = [
|
||||
'importlib-metadata; python_version<"3.9"',
|
||||
'packaging'
|
||||
'packaging',
|
||||
'frozendict',
|
||||
'typeguard'
|
||||
]
|
||||
dynamic = ["version"]
|
||||
|
||||
@@ -78,7 +80,7 @@ test = ["chacha_cicd_helper"]
|
||||
coverage-check = ["chacha_cicd_helper"]
|
||||
complexity-check = ["chacha_cicd_helper"]
|
||||
quality-check = ["chacha_cicd_helper"]
|
||||
type-check = ["chacha_cicd_helper"]
|
||||
type-check = ["chacha_cicd_helper","types-pytz"]
|
||||
doc-gen = ["chacha_cicd_helper"]
|
||||
|
||||
# [project.scripts]
|
||||
|
||||
@@ -11,4 +11,24 @@ Main module __init__ file.
|
||||
"""
|
||||
|
||||
from .__metadata__ import __version__, __Summuary__, __Name__
|
||||
from .test_module import test_function
|
||||
from .model import (
|
||||
DABFieldInfo,
|
||||
DABField,
|
||||
BaseAppliance,
|
||||
BaseFeature,
|
||||
DABModelException,
|
||||
MultipleInheritanceForbidden,
|
||||
BrokenInheritance,
|
||||
ReadOnlyField,
|
||||
NewFieldForbidden,
|
||||
NotAnnotatedField,
|
||||
ReadOnlyFieldAnnotation,
|
||||
InvalidFieldValue,
|
||||
InvalidFieldAnnotation,
|
||||
IncompletelyAnnotatedField,
|
||||
ImportForbidden,
|
||||
FunctionForbidden,
|
||||
FrozenDABField,
|
||||
InvalidFeatureInheritance,
|
||||
FeatureNotBound,
|
||||
)
|
||||
|
||||
797
src/dabmodel/model.py
Normal file
797
src/dabmodel/model.py
Normal file
@@ -0,0 +1,797 @@
|
||||
""" 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,
|
||||
List,
|
||||
Dict,
|
||||
Any,
|
||||
Tuple,
|
||||
Set,
|
||||
Annotated,
|
||||
FrozenSet,
|
||||
Callable,
|
||||
Type,
|
||||
)
|
||||
from types import UnionType, FunctionType, SimpleNamespace, MethodType
|
||||
from copy import deepcopy, copy
|
||||
|
||||
# from pprint import pprint
|
||||
import math
|
||||
|
||||
from frozendict import deepfreeze
|
||||
from typeguard import check_type, TypeCheckError, CollectionCheckStrategy
|
||||
|
||||
ALLOWED_ANNOTATIONS = {
|
||||
"Union": Union,
|
||||
"Optional": Optional,
|
||||
"List": List,
|
||||
"Dict": Dict,
|
||||
"Tuple": Tuple,
|
||||
"Set": Set,
|
||||
"FrozenSet": FrozenSet,
|
||||
"Annotated": Annotated,
|
||||
# builtins:
|
||||
"int": int,
|
||||
"str": str,
|
||||
"float": float,
|
||||
"bool": bool,
|
||||
"complex": complex,
|
||||
"bytes": bytes,
|
||||
"None": type(None),
|
||||
"list": list,
|
||||
"dict": dict,
|
||||
"set": set,
|
||||
"frozenset": frozenset,
|
||||
"tuple": tuple,
|
||||
}
|
||||
|
||||
ALLOWED_MODEL_FIELDS_TYPES = (str, int, float, complex, bool, bytes)
|
||||
|
||||
|
||||
# TV_ALLOWED_MODEL_FIELDS_TYPES = TypeVar("TV_ALLOWED_MODEL_FIELDS_TYPES", *ALLOWED_MODEL_FIELDS_TYPES, *ALLOWED_MODEL_FIELDS_CONTAINERS)
|
||||
|
||||
|
||||
class DABModelException(Exception):
|
||||
"""DABModelException Exception class
|
||||
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
|
||||
"""
|
||||
|
||||
|
||||
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 FunctionForbidden(DABModelException):
|
||||
"""FunctionForbidden Exception class
|
||||
function call are forbidden
|
||||
"""
|
||||
|
||||
class FeatureNotBound(DABModelException):
|
||||
"""FeatureNotBound Exception class
|
||||
a Feature must be bound to an Appliance
|
||||
"""
|
||||
|
||||
ALLOWED_HELPERS_MATH = SimpleNamespace(
|
||||
sqrt=math.sqrt,
|
||||
floor=math.floor,
|
||||
ceil=math.ceil,
|
||||
trunc=math.trunc,
|
||||
fabs=math.fabs,
|
||||
copysign=math.copysign,
|
||||
hypot=math.hypot,
|
||||
exp=math.exp,
|
||||
log=math.log,
|
||||
log10=math.log10,
|
||||
sin=math.sin,
|
||||
cos=math.cos,
|
||||
tan=math.tan,
|
||||
atan2=math.atan2,
|
||||
radians=math.radians,
|
||||
degrees=math.degrees,
|
||||
)
|
||||
|
||||
ALLOWED_HELPERS_DEFAULT = {
|
||||
"math": ALLOWED_HELPERS_MATH,
|
||||
"print": print,
|
||||
# Numbers & reducers (pure, deterministic)
|
||||
"abs": abs,
|
||||
"round": round,
|
||||
"min": min,
|
||||
"max": max,
|
||||
"sum": sum,
|
||||
# Introspection-free basics
|
||||
"len": len,
|
||||
"sorted": sorted,
|
||||
# Basic constructors (for copy-on-write patterns)
|
||||
"tuple": tuple,
|
||||
"list": list,
|
||||
"dict": dict,
|
||||
"set": set,
|
||||
# Simple casts if they need to normalize types
|
||||
"int": int,
|
||||
"float": float,
|
||||
"str": str,
|
||||
"bool": bool,
|
||||
"bytes": bytes,
|
||||
"complex": complex,
|
||||
# Easy iteration helpers (optional but handy)
|
||||
"range": range,
|
||||
}
|
||||
|
||||
|
||||
def _blocked_import(*args, **kwargs):
|
||||
raise ImportForbidden("imports disabled in __initializer")
|
||||
|
||||
|
||||
def _resolve_annotation(ann):
|
||||
if isinstance(ann, str):
|
||||
# Safe eval against a **whitelist** only
|
||||
return eval(ann, {"__builtins__": {}}, ALLOWED_ANNOTATIONS) # pylint: disable=eval-used
|
||||
return ann
|
||||
|
||||
|
||||
def _peel_annotated(t: Any) -> Any:
|
||||
# If you ever allow Annotated[T, ...], peel to T
|
||||
while True:
|
||||
origin = get_origin(t)
|
||||
if origin is None:
|
||||
return t
|
||||
name = getattr(origin, "__name__", "") or getattr(origin, "__qualname__", "") or str(origin)
|
||||
if "Annotated" in name:
|
||||
args = get_args(t)
|
||||
t = args[0] if args else t
|
||||
else:
|
||||
return t
|
||||
|
||||
|
||||
def _check_annotation_definition(_type) -> bool: # pylint: disable=too-complex,too-many-return-statements
|
||||
_type = _peel_annotated(_type)
|
||||
# handle Optional[] and Union[None,...]
|
||||
if (get_origin(_type) is Union or get_origin(_type) is UnionType) and type(None) in get_args(_type):
|
||||
return all(_check_annotation_definition(_) for _ in get_args(_type) if _ is not type(None))
|
||||
|
||||
# handle other Union[...]
|
||||
if get_origin(_type) is Union or get_origin(_type) is UnionType:
|
||||
return all(_check_annotation_definition(_) for _ in get_args(_type))
|
||||
|
||||
# handle Dict[...]
|
||||
if get_origin(_type) is dict:
|
||||
inner = get_args(_type)
|
||||
if len(inner) != 2:
|
||||
raise IncompletelyAnnotatedField(f"Dict Annotation requires 2 inner definitions: {_type}")
|
||||
return _peel_annotated(inner[0]) in ALLOWED_MODEL_FIELDS_TYPES and _check_annotation_definition(inner[1])
|
||||
|
||||
# handle Tuple[]
|
||||
if get_origin(_type) in [tuple]:
|
||||
inner_types = get_args(_type)
|
||||
if len(inner_types) == 0:
|
||||
raise IncompletelyAnnotatedField(f"Annotation requires inner definition: {_type}")
|
||||
if len(inner_types) == 2 and inner_types[1] is Ellipsis:
|
||||
return _check_annotation_definition(inner_types[0])
|
||||
return all(_check_annotation_definition(_) for _ in inner_types)
|
||||
|
||||
# handle Set[],Tuple[],FrozenSet[],List[]
|
||||
if get_origin(_type) in [set, frozenset, tuple, list]:
|
||||
inner_types = get_args(_type)
|
||||
if len(inner_types) == 0:
|
||||
raise IncompletelyAnnotatedField(f"Annotation requires inner definition: {_type}")
|
||||
return all(_check_annotation_definition(_) for _ in inner_types)
|
||||
|
||||
if _type in ALLOWED_MODEL_FIELDS_TYPES:
|
||||
return True
|
||||
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"""
|
||||
|
||||
__slots__ = ("_vals", "_types", "_touched", "_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, "_name", name)
|
||||
object.__setattr__(self, "_module", module)
|
||||
|
||||
@property
|
||||
def __name__(self) -> str:
|
||||
"""returns proxified class' name"""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def __module__(self) -> str:
|
||||
"""returns proxified module's name"""
|
||||
return self._module
|
||||
|
||||
@__module__.setter
|
||||
def __module__(self, value: str):
|
||||
pass
|
||||
|
||||
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: str, value: Any):
|
||||
"""internal proxy setattr"""
|
||||
if name not in self._types:
|
||||
raise NonExistingField(f"Cannot set unknown field {name}")
|
||||
T = self._types[name]
|
||||
|
||||
try:
|
||||
check_type(
|
||||
value,
|
||||
T,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(f"Field <{name}> value is not of expected type {T}.") from exp
|
||||
|
||||
self._vals[name] = value
|
||||
|
||||
def export(self) -> dict:
|
||||
"""exports all proxified values"""
|
||||
return dict(self._vals)
|
||||
|
||||
|
||||
T_Meta = TypeVar("T_Meta", bound="BaseMeta")
|
||||
T_BE = TypeVar("T_BE", bound="BaseElement")
|
||||
|
||||
|
||||
class BaseMeta(type):
|
||||
"""metaclass to use to build BaseElement"""
|
||||
|
||||
modified_field: Dict[str, Any] = {}
|
||||
new_fields: Dict[str, DABField[Any]] = {}
|
||||
initializer: Optional[Callable[..., Any]] = None
|
||||
__DABSchema__: dict[str, Any] = {}
|
||||
|
||||
@classmethod
|
||||
def pre_check(
|
||||
mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions : dict[str,Any]
|
||||
) -> None:
|
||||
"""early BaseElement checks"""
|
||||
# print("__NEW__ Defining:", name, "with keys:", list(namespace))
|
||||
|
||||
if len(bases) > 1:
|
||||
raise MultipleInheritanceForbidden("Multiple inheritance is not supported by dabmodel")
|
||||
if len(bases) == 0: # base class (BaseElement)
|
||||
namespace["__DABSchema__"] = {}
|
||||
else: # standard inheritance
|
||||
# check class tree origin
|
||||
if "__DABSchema__" not in dir(bases[0]):
|
||||
raise BrokenInheritance("__DABSchema__ not found in base class, broken inheritance chain.")
|
||||
# copy inherited schema
|
||||
namespace["__DABSchema__"] = copy(bases[0].__DABSchema__)
|
||||
|
||||
# force field without default value to be instantiated (with None)
|
||||
if "__annotations__" in namespace:
|
||||
for _funknown in [_ for _ in namespace["__annotations__"] if _ not in namespace.keys()]:
|
||||
namespace[_funknown] = None
|
||||
|
||||
@classmethod
|
||||
def pre_processing(mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any],extensions : 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
|
||||
elif isinstance(_fvalue, classmethod):
|
||||
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, extensions)
|
||||
else: # New fieds
|
||||
mcs.pre_processing_new_fields(name, bases, namespace, _fname, _fvalue, extensions)
|
||||
# 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,
|
||||
extensions : dict[str,Any]
|
||||
): # pylint: disable=unused-argument
|
||||
"""preprocessing BaseElement modified Fields"""
|
||||
if "__annotations__" in namespace and _fname in namespace["__annotations__"]:
|
||||
raise ReadOnlyFieldAnnotation(f"annotations cannot be modified on derived classes {_fname}")
|
||||
try:
|
||||
check_type(
|
||||
_fvalue,
|
||||
namespace["__DABSchema__"][_fname].annotations,
|
||||
collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS,
|
||||
)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
f"Field <{_fname}> New Field value is not of expected type {bases[0].__annotations__[_fname]}."
|
||||
) from exp
|
||||
mcs.modified_field[_fname] = _fvalue
|
||||
|
||||
@classmethod
|
||||
def pre_processing_new_fields(
|
||||
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
|
||||
"""preprocessing BaseElement new Fields"""
|
||||
#print(f"New field: {_fname}")
|
||||
|
||||
# check if field is annotated
|
||||
if "__annotations__" not in namespace or _fname not in namespace["__annotations__"]:
|
||||
raise NotAnnotatedField(f"Every dabmodel Fields must be annotated ({_fname})")
|
||||
|
||||
# check if annotation is allowed
|
||||
if isinstance(namespace["__annotations__"][_fname], str):
|
||||
namespace["__annotations__"][_fname] = _resolve_annotation(namespace["__annotations__"][_fname])
|
||||
|
||||
if not _check_annotation_definition(namespace["__annotations__"][_fname]):
|
||||
raise InvalidFieldAnnotation(f"Field <{_fname}> has not an allowed or valid annotation.")
|
||||
|
||||
_finfo: DABFieldInfo = DABFieldInfo()
|
||||
origin = get_origin(namespace["__annotations__"][_fname])
|
||||
tname = getattr(origin, "__name__", "") or getattr(origin, "__qualname__", "") or str(origin)
|
||||
if "Annotated" in tname:
|
||||
args = get_args(namespace["__annotations__"][_fname])
|
||||
if args:
|
||||
if len(args) > 2:
|
||||
raise InvalidFieldAnnotation(f"Field <{_fname}> had invalid Annotated value.")
|
||||
if len(args) == 2 and not isinstance(args[1], DABFieldInfo):
|
||||
raise InvalidFieldAnnotation("Only DABFieldInfo object is allowed as Annotated data.")
|
||||
|
||||
_finfo = args[1]
|
||||
|
||||
# check if value is valid
|
||||
try:
|
||||
check_type(_fvalue, namespace["__annotations__"][_fname], collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS)
|
||||
except TypeCheckError as exp:
|
||||
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 call_initializer(
|
||||
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions : dict[str,Any]
|
||||
):
|
||||
"""BaseElement initializer processing"""
|
||||
if mcs.initializer is not None:
|
||||
init_fieldvalues = {}
|
||||
init_fieldtypes = {}
|
||||
for _fname, _fvalue in cls.__DABSchema__.items():
|
||||
if isinstance(_fvalue,DABField):
|
||||
init_fieldvalues[_fname] = deepcopy(_fvalue.raw_value)
|
||||
init_fieldtypes[_fname] = _fvalue.annotations
|
||||
fakecls = ModelSpecView(init_fieldvalues, init_fieldtypes, cls.__name__, cls.__module__)
|
||||
safe_globals = {"__builtins__": {"__import__": _blocked_import}, **ALLOWED_HELPERS_DEFAULT}
|
||||
if mcs.initializer.__code__.co_freevars:
|
||||
raise FunctionForbidden("__initializer must not use closures")
|
||||
safe_initializer = FunctionType(
|
||||
mcs.initializer.__code__,
|
||||
safe_globals,
|
||||
name=mcs.initializer.__name__,
|
||||
argdefs=mcs.initializer.__defaults__,
|
||||
closure=None,
|
||||
)
|
||||
safe_initializer(fakecls) # pylint: disable=not-callable
|
||||
for _fname, _fvalue in fakecls.export().items():
|
||||
try:
|
||||
check_type(_fvalue, cls.__DABSchema__[_fname].annotations, collection_check_strategy=CollectionCheckStrategy.ALL_ITEMS)
|
||||
except TypeCheckError as exp:
|
||||
raise InvalidFieldValue(
|
||||
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"""
|
||||
extensions : dict[str,Any] = dict()
|
||||
mcs.pre_check(name, bases, namespace, extensions)
|
||||
mcs.pre_processing(name, bases, namespace, extensions)
|
||||
|
||||
_cls = super().__new__(mcs, name, bases, namespace)
|
||||
|
||||
mcs.save_values(_cls, name, bases, namespace, extensions)
|
||||
mcs.call_initializer(_cls, name, bases, namespace, extensions)
|
||||
_cls.install_guard(extensions)
|
||||
|
||||
return _cls
|
||||
|
||||
@classmethod
|
||||
def save_values(
|
||||
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions : dict[str,Any]
|
||||
):
|
||||
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: Type, *args: Any, **kw: Any): # intentionally untyped
|
||||
"""BaseElement new instance"""
|
||||
obj = super().__call__(*args, **kw)
|
||||
|
||||
extensions = dict()
|
||||
|
||||
for _fname, _fvalue in cls.__DABSchema__.items():
|
||||
if isinstance(_fvalue,DABField):
|
||||
object.__setattr__(obj, _fname, _fvalue.value)
|
||||
|
||||
inst_schema = copy(obj.__DABSchema__)
|
||||
for _fname, _fvalue in cls.__DABSchema__.items():
|
||||
if isinstance(_fvalue,DABField):
|
||||
inst_schema[_fname] = FrozenDABField(_fvalue)
|
||||
|
||||
object.__setattr__(obj, "__DABSchema__", inst_schema)
|
||||
cls.modify_object(obj,extensions)
|
||||
|
||||
|
||||
return obj
|
||||
|
||||
def modify_object(cls:Type,obj,extensions : dict[str,Any]):
|
||||
pass
|
||||
|
||||
|
||||
def install_guard(cls:Type,extensions : dict[str,Any]):
|
||||
orig_setattr = getattr(cls,"__setattr__")
|
||||
|
||||
# cls.orig_setattr = orig_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 key in _self.__dict__:
|
||||
raise ReadOnlyField(f"{key} is read-only")
|
||||
elif key in _self.__DABSchema__["features"].keys():
|
||||
if key in _self.__dict__:
|
||||
raise ReadOnlyField(f"{key} is read-only")
|
||||
else:
|
||||
raise NewFieldForbidden("creating new fields is not allowed")
|
||||
|
||||
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):
|
||||
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):
|
||||
|
||||
@classmethod
|
||||
def pre_check(
|
||||
mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions : dict[str,Any]
|
||||
) -> None:
|
||||
"""early BaseElement checks"""
|
||||
super().pre_check(name,bases,namespace,extensions)
|
||||
if "features" not in namespace["__DABSchema__"]:
|
||||
namespace["__DABSchema__"]["features"]={}
|
||||
else:
|
||||
namespace["__DABSchema__"]["features"] = copy(namespace["__DABSchema__"]["features"])
|
||||
return
|
||||
|
||||
|
||||
@classmethod
|
||||
def pre_processing(mcs: type["BaseMeta"], name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any],extensions : dict[str,Any]):
|
||||
extensions["new_features"]: dict[str,type[BaseFeature]] = {}
|
||||
extensions["modified_features"]: dict[str,type[BaseFeature]] = {}
|
||||
super().pre_processing(name,bases,namespace,extensions)
|
||||
|
||||
|
||||
@classmethod
|
||||
def pre_processing_new_fields(
|
||||
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
|
||||
"""preprocessing BaseElement new Fields"""
|
||||
#print('pre_processing_new_fields')
|
||||
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,BaseMetaFeature):
|
||||
extensions["new_features"][_fname]=_fvalue
|
||||
else:
|
||||
super().pre_processing_new_fields(name,bases,namespace,_fname,_fvalue,extensions)
|
||||
|
||||
@classmethod
|
||||
def save_values(
|
||||
mcs: type["BaseMeta"], cls, name: str, bases: tuple[type[Any], ...], namespace: dict[str, Any], # pylint: disable=unused-argument
|
||||
extensions : dict[str,Any]
|
||||
):
|
||||
super().save_values(cls,name,bases,namespace,extensions)
|
||||
|
||||
for _ftname,_ftvalue in extensions["modified_features"].items():
|
||||
_ftvalue._BoundAppliance = cls
|
||||
cls.__DABSchema__["features"][_ftname] = _ftvalue
|
||||
for _ftname,_ftvalue in extensions["new_features"].items():
|
||||
_ftvalue._BoundAppliance = cls
|
||||
cls.__DABSchema__["features"][_ftname] = _ftvalue
|
||||
|
||||
def modify_object(cls:Type, obj, extensions : dict[str,Any]): # intentionally untyped
|
||||
for _ftname,_ftvalue in cls.__DABSchema__["features"].items():
|
||||
instft = _ftvalue()
|
||||
object.__setattr__(obj, _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.
|
||||
"""
|
||||
@@ -1,43 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# dabmodel (c) by chacha
|
||||
#
|
||||
# dabmodel is licensed under a
|
||||
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
||||
#
|
||||
# You should have received a copy of the license along with this
|
||||
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
||||
|
||||
"""Phasellus tellus lectus, volutpat eu dapibus ut, suscipit vel augue.
|
||||
|
||||
Tips:
|
||||
Aliquam non leo vel libero sagittis viverra. Quisque lobortis nunc sit amet augue euismod laoreet.
|
||||
Note:
|
||||
Maecenas volutpat porttitor pretium. Aliquam suscipit quis nisi non imperdiet.
|
||||
Note:
|
||||
Vivamus et efficitur lorem, eget imperdiet tortor. Integer vel interdum sem.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING: # Only imports the below statements during type checking
|
||||
pass
|
||||
|
||||
def test_function(testvar: int) -> int:
|
||||
""" A test function that return testvar+1 and print "Hello world !"
|
||||
|
||||
Proin eget sapien eget ipsum efficitur mollis nec ac nibh.
|
||||
|
||||
Note:
|
||||
Morbi id lectus maximus, condimentum nunc eget, porta felis. In tristique velit tortor.
|
||||
|
||||
Args:
|
||||
testvar: any integer
|
||||
|
||||
Returns:
|
||||
testvar+1
|
||||
"""
|
||||
print("Hello world !")
|
||||
return testvar+1
|
||||
17
src/dabmodel/tools.py
Normal file
17
src/dabmodel/tools.py
Normal file
@@ -0,0 +1,17 @@
|
||||
"""library's internal tools"""
|
||||
|
||||
from uuid import UUID
|
||||
from datetime import datetime
|
||||
import json
|
||||
|
||||
|
||||
class DABJSONEncoder(json.JSONEncoder):
|
||||
"""allows to JSON encode non supported data type"""
|
||||
|
||||
def default(self, o):
|
||||
if isinstance(o, UUID):
|
||||
# if the o is uuid, we simply return the value of uuid
|
||||
return o.hex
|
||||
if isinstance(o, datetime):
|
||||
return str(o)
|
||||
return json.JSONEncoder.default(self, o)
|
||||
1360
test/test_model.py
Normal file
1360
test/test_model.py
Normal file
File diff suppressed because it is too large
Load Diff
@@ -1,35 +0,0 @@
|
||||
# dabmodel (c) by chacha
|
||||
#
|
||||
# dabmodel is licensed under a
|
||||
# Creative Commons Attribution-NonCommercial-ShareAlike 4.0 International Unported License.
|
||||
#
|
||||
# You should have received a copy of the license along with this
|
||||
# work. If not, see <https://creativecommons.org/licenses/by-nc-sa/4.0/>.
|
||||
|
||||
import unittest
|
||||
from os import chdir
|
||||
from io import StringIO
|
||||
from contextlib import redirect_stdout,redirect_stderr
|
||||
from pathlib import Path
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
|
||||
from src import dabmodel
|
||||
|
||||
testdir_path = Path(__file__).parent.resolve()
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
class Testtest_module(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
def test_version(self):
|
||||
self.assertNotEqual(dabmodel.__version__,"?.?.?")
|
||||
|
||||
def test_test_module(self):
|
||||
|
||||
with redirect_stdout(StringIO()) as capted_stdout, redirect_stderr(StringIO()) as capted_stderr:
|
||||
self.assertEqual(dabmodel.test_function(41),42)
|
||||
self.assertEqual(len(capted_stderr.getvalue()),0)
|
||||
self.assertEqual(capted_stdout.getvalue().strip(),"Hello world !")
|
||||
Reference in New Issue
Block a user