good progress
This commit is contained in:
@@ -2,20 +2,52 @@ from __future__ import annotations
|
||||
|
||||
from abc import ABC, ABCMeta
|
||||
from uuid import uuid4
|
||||
from typing import Annotated, ClassVar, Any, Callable, Self, TypeVar, TypeAlias
|
||||
from typing_extensions import dataclass_transform
|
||||
from pydantic import BaseModel, StrictInt, StrictStr, constr, ByteSize, AwareDatetime, UUID4, model_validator
|
||||
from typing import Annotated, ClassVar, Any, Self, TypeVar, TypeAlias, Generic
|
||||
from datetime import datetime
|
||||
from copy import deepcopy, copy
|
||||
from typing_extensions import dataclass_transform, get_origin
|
||||
from pydantic import (
|
||||
ConfigDict,
|
||||
BaseModel,
|
||||
StrictInt,
|
||||
StrictStr,
|
||||
constr,
|
||||
ByteSize,
|
||||
AwareDatetime,
|
||||
UUID4,
|
||||
model_validator,
|
||||
field_validator,
|
||||
field_serializer,
|
||||
SerializeAsAny,
|
||||
)
|
||||
from pydantic.fields import Field, _Unset, PydanticUndefined
|
||||
from pydantic._internal._model_construction import ModelMetaclass, PydanticModelField
|
||||
from pydantic._internal._generics import PydanticGenericMetadata
|
||||
from pydantic._internal._decorators import ensure_classmethod_based_on_signature
|
||||
from datetime import datetime
|
||||
import pytz
|
||||
from copy import deepcopy
|
||||
|
||||
from runtype import issubclass as runtype_issubclass
|
||||
|
||||
|
||||
class NoInstanceMethod:
|
||||
"""Descriptor to forbid that other descriptors can be looked up on an instance"""
|
||||
|
||||
def __init__(self, descr, name=None):
|
||||
self.descr = descr
|
||||
self.name = name
|
||||
|
||||
def __set_name__(self, owner, name):
|
||||
self.name = name
|
||||
|
||||
def __get__(self, instance, owner):
|
||||
# enforce the instance cannot look up the attribute at all
|
||||
if instance is not None:
|
||||
raise AttributeError(f"{type(instance).__name__!r} has no attribute {self.name!r}")
|
||||
# invoke any descriptor we are wrapping
|
||||
return self.descr.__get__(instance, owner)
|
||||
|
||||
|
||||
def DABField(default: Any = PydanticUndefined, *, final: bool | None = _Unset, finalize_only: bool | None = _Unset, **kwargs):
|
||||
kwargs["frozen"] = True
|
||||
return Field(default, **kwargs)
|
||||
|
||||
|
||||
@@ -26,12 +58,21 @@ T_BaseElement_ConfigMethod_OrderedSet: TypeAlias = dict[T_BaseElement_ConfigMeth
|
||||
|
||||
|
||||
class ConfigElement:
|
||||
default_values_override_methods: T_BaseElement_ConfigMethod_OrderedSet = {}
|
||||
main_build_method: T_BaseElement_ConfigMethod_OrderedSet = {}
|
||||
def __init__(self) -> None:
|
||||
self.default_values_override_methods: T_BaseElement_ConfigMethod_OrderedSet = {}
|
||||
self.main_build_method: T_BaseElement_ConfigMethod_OrderedSet = {}
|
||||
|
||||
def __copy__(self) -> Self:
|
||||
# we cannot seepcopy because of classmethods, so we do a copy++
|
||||
cls = self.__class__
|
||||
result = cls.__new__(cls)
|
||||
for k, v in self.__dict__.items():
|
||||
setattr(result, k, copy(v))
|
||||
return result
|
||||
|
||||
|
||||
class IBaseElement(ABC):
|
||||
_config_element: ConfigElement = ConfigElement()
|
||||
class IBaseElement(BaseModel, ABC):
|
||||
_config_element: ClassVar[ConfigElement] = ConfigElement()
|
||||
|
||||
|
||||
@dataclass_transform(kw_only_default=True, field_specifiers=(PydanticModelField,))
|
||||
@@ -56,50 +97,74 @@ class BaseElementMeta(ModelMetaclass, ABCMeta):
|
||||
_create_model_module,
|
||||
**kwargs,
|
||||
)
|
||||
# print(result)
|
||||
# print(type(result))
|
||||
assert issubclass(result, IBaseElement)
|
||||
# copy base class default-configs
|
||||
if not "_config_element" in result.__dict__:
|
||||
# print(result.__base__)
|
||||
# print(type(result.__base__))
|
||||
assert result.__base__ is not None
|
||||
print(cls_name)
|
||||
|
||||
assert issubclass(result, IBaseElement), "Only IBaseElement subclasses are supported"
|
||||
|
||||
# forcing all Fields to be frozen
|
||||
for _, field_val in result.model_fields.items():
|
||||
field_val.frozen = True
|
||||
|
||||
# copying/forwarding base classes default-configs
|
||||
if "_config_element" not in result.__dict__:
|
||||
assert result.__base__ is not None, "Only IBaseElement subclasses are supported"
|
||||
if issubclass(result.__base__, IBaseElement):
|
||||
result._config_element = deepcopy(result.__base__._config_element)
|
||||
result._config_element = copy(result.__base__._config_element)
|
||||
else:
|
||||
result._config_element = ConfigElement()
|
||||
|
||||
# searching and storing current class default-configs
|
||||
for _, method in result.__dict__.items():
|
||||
if isinstance(method, classmethod):
|
||||
if hasattr(method, "default_values_override"):
|
||||
result._config_element.default_values_override_methods[method] = None
|
||||
|
||||
# todo: find a way to 'lock' and add restriction to a field after inheritance
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class BaseElement(
|
||||
IBaseElement,
|
||||
BaseModel,
|
||||
ABC,
|
||||
validate_assignment=True,
|
||||
revalidate_instances="subclass-instances",
|
||||
validate_default=True,
|
||||
metaclass=BaseElementMeta,
|
||||
):
|
||||
class Config:
|
||||
ignored_types = (NoInstanceMethod,)
|
||||
|
||||
template_id: Annotated[UUID4, DABField(..., repr=True)]
|
||||
template_short_name: Annotated[
|
||||
StrictStr, constr(strip_whitespace=True, to_lower=True, strict=True, max_length=16), DABField(..., repr=True)
|
||||
]
|
||||
template_long_name: Annotated[StrictStr | None, DABField()]
|
||||
template_description: Annotated[StrictStr | None, DABField()]
|
||||
_saved_default_value: ClassVar[dict[str, Any]]
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def __default_values_override_hook__(cls, values: T_BaseElement_ConfigMethod_Arg) -> T_BaseElement_ConfigMethod_Arg:
|
||||
new_values = deepcopy(values)
|
||||
# extracting default values that were set in model fields
|
||||
cls._saved_default_value = dict()
|
||||
for field_key, field_val in cls.model_fields.items():
|
||||
assert field_val.annotation is not None, "all fields must have annotation"
|
||||
assert not runtype_issubclass(
|
||||
field_val.annotation, BaseFeature
|
||||
), "Features can only be in Appliance's features[] dict attribute"
|
||||
if field_val.default != PydanticUndefined:
|
||||
cls._saved_default_value[field_key] = deepcopy(field_val.default)
|
||||
|
||||
# walking through each default_values_override functions
|
||||
for method, _ in cls._config_element.default_values_override_methods.items():
|
||||
new_values = method.__func__(cls, new_values)
|
||||
method.__func__(cls, cls._saved_default_value)
|
||||
|
||||
# applying user-defined values
|
||||
for key, _ in values.items():
|
||||
new_values[key] = values[key]
|
||||
return new_values
|
||||
cls._saved_default_value[key] = values[key]
|
||||
|
||||
return cls._saved_default_value
|
||||
|
||||
|
||||
def default_values_override(func: T_BaseElement_ConfigMethod) -> T_BaseElement_ConfigMethod:
|
||||
@@ -108,14 +173,24 @@ def default_values_override(func: T_BaseElement_ConfigMethod) -> T_BaseElement_C
|
||||
return func
|
||||
|
||||
|
||||
class BaseFeature(BaseElement, ABC):
|
||||
compatible_appliance: Annotated[BaseAppliance, DABField()]
|
||||
class BaseFeature(BaseElement, ABC): ...
|
||||
|
||||
|
||||
# FeatureSet = dict[BaseFeature, None]
|
||||
T_Feature = TypeVar("T_Feature", bound=BaseFeature)
|
||||
T_BaseAppliance_Feature_OrderedSet: TypeAlias = dict[type[T_Feature], T_Feature]
|
||||
|
||||
|
||||
class MySet(set):
|
||||
def __init__(self, _iter, klass=None):
|
||||
if klass is not None:
|
||||
for item in iter:
|
||||
if not isinstance(item, klass):
|
||||
raise Exception("Error")
|
||||
super(MySet, self).__init__(_iter)
|
||||
|
||||
|
||||
class BaseAppliance(BaseElement, ABC):
|
||||
|
||||
cpu_cnt: Annotated[StrictInt, DABField(1, gt=0)]
|
||||
ram_size: Annotated[ByteSize, DABField(256, gt=128)]
|
||||
swap_size: Annotated[ByteSize, DABField(200, ge=0)]
|
||||
@@ -130,12 +205,26 @@ class BaseAppliance(BaseElement, ABC):
|
||||
dabinst_description: Annotated[StrictStr | None, DABField("")]
|
||||
dabinst_creationdate: Annotated[AwareDatetime | None, DABField(datetime.now(tz=pytz.utc))]
|
||||
|
||||
# __features: ClassVar[FeatureSet] = []
|
||||
features: Annotated[SerializeAsAny[T_BaseAppliance_Feature_OrderedSet], DABField({})]
|
||||
|
||||
def _add_feature(self, feature: BaseFeature) -> None: ...
|
||||
@field_validator("features", mode="after")
|
||||
@classmethod
|
||||
def serialize_features(cls, features: T_BaseAppliance_Feature_OrderedSet) -> dict[str, BaseFeature]:
|
||||
return {str(key.__name__): value for key, value in features.items()}
|
||||
|
||||
# self.__features.append(feature)
|
||||
@NoInstanceMethod
|
||||
@classmethod
|
||||
def add_feature(cls, feat: BaseFeature):
|
||||
print("Addfeature")
|
||||
print(cls._saved_default_value["features"])
|
||||
cls._saved_default_value["features"][type(feat)] = feat
|
||||
|
||||
@NoInstanceMethod
|
||||
@classmethod
|
||||
def del_feature(cls, type_feat: type[BaseFeature]):
|
||||
del cls._saved_default_value["features"][type_feat]
|
||||
|
||||
# def __init__(self, *args, **data):
|
||||
# super().__init__(*args, **data)
|
||||
@NoInstanceMethod
|
||||
@classmethod
|
||||
def get_feature(cls, type_feat: type[T_Feature]) -> T_Feature:
|
||||
return cls._saved_default_value["features"][type_feat]
|
||||
|
||||
@@ -11,6 +11,7 @@ from os import chdir
|
||||
from pathlib import Path
|
||||
|
||||
from pydantic import StrictInt, model_validator
|
||||
from pydantic.fields import Field
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
@@ -24,44 +25,83 @@ chdir(testdir_path.parent.resolve())
|
||||
|
||||
|
||||
class MyAppliance(dabmodel.BaseAppliance):
|
||||
|
||||
app_specifi_integer_arg: Annotated[StrictInt | None, dabmodel.DABField(42)]
|
||||
|
||||
class MyFeature(dabmodel.BaseFeature):
|
||||
@dabmodel.default_values_override
|
||||
@classmethod
|
||||
def __override_config__(cls, values):
|
||||
print("!!! CONFIG Feature 1")
|
||||
values["template_id"] = "421d61cb-e364-46d8-9b77-ec439f1fa666"
|
||||
values["template_short_name"] = "my-feature-1"
|
||||
values["template_long_name"] = "My feature template 1 !!"
|
||||
values["template_description"] = """A very nice FEature 1"""
|
||||
|
||||
class MyFeature2(dabmodel.BaseFeature):
|
||||
@dabmodel.default_values_override
|
||||
@classmethod
|
||||
def __override_config__(cls, values):
|
||||
print("!!! CONFIG Feature 2")
|
||||
values["template_id"] = "421d61cb-e364-46d8-9b77-ec439f1fa666"
|
||||
values["template_short_name"] = "my-feature-2"
|
||||
values["template_long_name"] = "My feature template 2 !!"
|
||||
values["template_description"] = """A very nice FEature 2"""
|
||||
|
||||
@dabmodel.default_values_override
|
||||
@classmethod
|
||||
def __override_config__(cls, values):
|
||||
print("!!! CONFIG 1")
|
||||
print("!!! CONFIG Appliance 1")
|
||||
print(f"!!!! {values['rootfs_size']}")
|
||||
values["template_id"] = "421d61cb-e364-46d8-9b64-ec439f1faae8"
|
||||
values["template_short_name"] = "my-app- tem 1"
|
||||
values["template_long_name"] = "My appliance template 1 !!"
|
||||
values["template_description"] = """A very nice Appliance 1"""
|
||||
values["ram_size"] = 1024
|
||||
return values
|
||||
# values["features"][cls.MyFeature] = MyAppliance.MyFeature()
|
||||
# values["features"][cls.MyFeature2] = MyAppliance.MyFeature2()
|
||||
cls.add_feature(MyAppliance.MyFeature())
|
||||
cls.add_feature(MyAppliance.MyFeature2())
|
||||
|
||||
|
||||
class MyAppliance2(MyAppliance):
|
||||
@dabmodel.default_values_override
|
||||
@classmethod
|
||||
def __override_config__(cls, values):
|
||||
print("!!! CONFIG 2")
|
||||
print("!!! CONFIG Appliance 2")
|
||||
print(f"!!!! {values['template_id']}")
|
||||
values["template_id"] = "421d61cb-e664-46d8-9b64-ec439f1fafff"
|
||||
values["template_short_name"] = "my-app- tem 2"
|
||||
values["template_long_name"] = "My appliance template 2 !!"
|
||||
values["template_description"] = """A very nice Appliance 2"""
|
||||
return values
|
||||
# del values["features"][MyAppliance.MyFeature]
|
||||
cls.del_feature(MyAppliance.MyFeature)
|
||||
|
||||
|
||||
class MyAppliance3(dabmodel.BaseAppliance):
|
||||
|
||||
class MyFeature(MyAppliance.MyFeature):
|
||||
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case
|
||||
|
||||
@dabmodel.default_values_override
|
||||
@classmethod
|
||||
def __override_config__(cls, values):
|
||||
print("!!! CONFIG Feature 1 (modified)")
|
||||
values["template_id"] = "421d61cb-e364-46d8-9b77-ec439f1fa777"
|
||||
values["template_short_name"] = "my-feature-1-bis"
|
||||
|
||||
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case
|
||||
|
||||
@dabmodel.default_values_override
|
||||
@classmethod
|
||||
def __override_config__(cls, values):
|
||||
print("!!! CONFIG 3")
|
||||
print("!!! CONFIG Appliance 3")
|
||||
values["template_id"] = "421d61cb-e364-46d8-9b64-ec439f1faaaa"
|
||||
values["template_short_name"] = "my-app- tem 3"
|
||||
values["template_long_name"] = "My appliance template 3 !!"
|
||||
values["template_description"] = """A very nice Appliance 3"""
|
||||
values["ram_size"] = 3076
|
||||
return values
|
||||
# values["features"][cls.MyFeature] = cls.MyFeature()
|
||||
cls.add_feature(cls.MyFeature())
|
||||
|
||||
|
||||
class TestModel(unittest.TestCase):
|
||||
@@ -73,14 +113,25 @@ class TestModel(unittest.TestCase):
|
||||
|
||||
def test_model(self):
|
||||
|
||||
feature1 = MyAppliance.MyFeature()
|
||||
print(feature1)
|
||||
print(MyAppliance.MyFeature)
|
||||
print(MyAppliance.MyFeature.__name__)
|
||||
print(MyAppliance.MyFeature.__class__)
|
||||
print("==")
|
||||
print(feature1.model_dump_json(indent=1))
|
||||
|
||||
app = MyAppliance(dabinst_short_name="my-app-1", app_specifi_integer_arg=123)
|
||||
# app.template_short_name = "tete !"
|
||||
print(app.model_dump_json(indent=1))
|
||||
|
||||
app2 = MyAppliance2(dabinst_short_name="my-app-2", app_specifi_integer_arg=654)
|
||||
# app.template_short_name = "tete !"
|
||||
print(app2.model_dump_json(indent=1))
|
||||
app3 = MyAppliance3(dabinst_short_name="my-app-3", template_description="FORCED")
|
||||
|
||||
app3 = MyAppliance3(dabinst_short_name="my-app-3", app_specifi_integer_arg=654, template_description="FORCED")
|
||||
# app.template_short_name = "tete !"
|
||||
print(app.model_dump_json(indent=1))
|
||||
print(app2.model_dump_json(indent=1))
|
||||
print(app3.model_dump_json(indent=1))
|
||||
|
||||
# app3.add_feature(MyAppliance.MyFeature())
|
||||
# app3.add_feature(MyAppliance.MyFeature())
|
||||
# app3.add_feature(MyAppliance.MyFeature())
|
||||
# print(app3.model_dump_json(indent=1))
|
||||
# app3 = MyAppliance3(dabinst_short_name="my-app-3", template_description="FORCED")
|
||||
# print(app3.model_dump_json(indent=1))
|
||||
|
||||
Reference in New Issue
Block a user