Compare commits

...

3 Commits

Author SHA1 Message Date
chacha
29827b51bc add chatgpt code :) 2025-08-31 22:19:58 +02:00
chacha
7440731135 dev 2025-08-31 21:23:03 +02:00
chacha
0eef35e36f continue work 2024-12-08 01:04:27 +01:00
6 changed files with 480 additions and 443 deletions

View File

@@ -1,13 +1,20 @@
<?xml version="1.0" encoding="UTF-8" standalone="no"?> <?xml version="1.0" encoding="UTF-8" standalone="no"?>
<?eclipse-pydev version="1.0"?><pydev_project> <?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_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"> <pydev_property name="org.python.pydev.PYTHON_PROJECT_VERSION">python interpreter</pydev_property>
<path>/${PROJECT_DIR_NAME}/src</path>
<path>/${PROJECT_DIR_NAME}</path>
</pydev_pathproperty> <pydev_pathproperty name="org.python.pydev.PROJECT_SOURCE_PATH">
</pydev_project> <path>/${PROJECT_DIR_NAME}/src</path>
<path>/${PROJECT_DIR_NAME}</path>
</pydev_pathproperty>
</pydev_project>

View File

@@ -34,7 +34,9 @@ classifiers = [
] ]
dependencies = [ dependencies = [
'importlib-metadata; python_version<"3.9"', 'importlib-metadata; python_version<"3.9"',
'packaging' 'packaging',
'pydantic',
'runtype'
] ]
dynamic = ["version"] dynamic = ["version"]
@@ -78,7 +80,7 @@ test = ["chacha_cicd_helper"]
coverage-check = ["chacha_cicd_helper"] coverage-check = ["chacha_cicd_helper"]
complexity-check = ["chacha_cicd_helper"] complexity-check = ["chacha_cicd_helper"]
quality-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"] doc-gen = ["chacha_cicd_helper"]
# [project.scripts] # [project.scripts]

View File

@@ -11,4 +11,4 @@ Main module __init__ file.
""" """
from .__metadata__ import __version__, __Summuary__, __Name__ from .__metadata__ import __version__, __Summuary__, __Name__
from .model import DABField, BaseFeature, BaseAppliance, default_values_override from .model import BaseFeature, BaseAppliance

View File

@@ -1,267 +1,383 @@
from __future__ import annotations from __future__ import annotations
from typing import Any, Dict, Optional, TypeVar, Generic, get_origin, get_args, Annotated as _Annotated, Union
from uuid import uuid4, UUID
from datetime import datetime as _dt
from abc import ABC, ABCMeta, abstractmethod
from uuid import uuid4
from typing import Annotated, ClassVar, Any, Self, TypeVar, TypeAlias, Generic, Union
from datetime import datetime
from copy import deepcopy, copy
from typing_extensions import dataclass_transform, get_origin
from pydantic import ( from pydantic import (
ConfigDict,
BaseModel, BaseModel,
StrictInt, Field,
StrictStr, ConfigDict,
constr, model_validator,
ByteSize,
AwareDatetime, AwareDatetime,
UUID4, UUID4,
model_validator, ByteSize,
field_validator, StrictInt,
field_serializer, StrictStr,
SerializeAsAny,
) )
from pydantic.fields import Field, _Unset, PydanticUndefined from pydantic._internal._model_construction import ModelMetaclass
from pydantic._internal._model_construction import ModelMetaclass, PydanticModelField
from pydantic._internal._generics import PydanticGenericMetadata
from pydantic._internal._decorators import ensure_classmethod_based_on_signature
import pytz import pytz
import inspect
from runtype import issubclass as runtype_issubclass
class NoInstanceMethod: # ===================== Maintainers policy =====================
"""Descriptor to forbid that other descriptors can be looked up on an instance"""
def __init__(self, descr, name=None): ALLOWED_APPLIANCE_FIELD_TYPES: set[type] = {
self.descr = descr StrictInt, StrictStr, ByteSize, UUID4, UUID, AwareDatetime, _dt, bool, int, str,
self.name = name }
ALLOWED_FEATURE_FIELD_TYPES: set[type] = {
def __set_name__(self, owner, name): StrictInt, StrictStr, ByteSize, UUID4, UUID, AwareDatetime, _dt, bool, int, str,
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): def _is_allowed_type(annotation: Any, allowed: set[type]) -> bool:
return Field(default, **kwargs) origin = get_origin(annotation)
if origin is _Annotated or (origin is not None and getattr(origin, "__name__", "") == "Annotated"):
return _is_allowed_type(get_args(annotation)[0], allowed)
if origin is None:
return annotation in allowed
if origin is Union:
args = [a for a in get_args(annotation) if a is not type(None)]
return all(_is_allowed_type(a, allowed) for a in args)
args = get_args(annotation)
if origin in (list, tuple, dict, set):
return all(_is_allowed_type(a, allowed) for a in args if a is not None)
return all(_is_allowed_type(a, allowed) for a in args if a is not None)
T_BaseElement = TypeVar("T_BaseElement", bound="BaseElement") def _is_feature_annotation(annotation: Any) -> bool:
T_BaseElement_ConfigMethod_Arg: TypeAlias = dict[str, Any] origin = get_origin(annotation)
T_BaseElement_ConfigMethod: TypeAlias = "classmethod[T_BaseElement, [T_BaseElement_ConfigMethod_Arg], T_BaseElement_ConfigMethod_Arg]" if origin is Union:
args = [a for a in get_args(annotation) if a is not type(None)]
return any(_is_feature_annotation(a) for a in args)
base = get_args(annotation)[0] if get_origin(annotation) is Optional else annotation
try:
return isinstance(base, type) and issubclass(base, BaseFeature) # type: ignore[name-defined]
except Exception:
return False
class ConfigElement: def _annotation_equal(a: Any, b: Any) -> bool:
def __init__(self) -> None: oa, ob = get_origin(a), get_origin(b)
self.default_values_override_methods: dict[T_BaseElement_ConfigMethod, None] = {} if oa is None and ob is None:
self.main_build_method: dict[T_BaseElement_ConfigMethod, None] = {} return a is b
if oa != ob:
def __copy__(self) -> Self: return False
# we cannot deepcopy because of classmethods, so we do a manual enhanced copy aa, ab = get_args(a), get_args(b)
cls = self.__class__ if len(aa) != len(ab):
result = cls.__new__(cls) return False
for k, v in self.__dict__.items(): return all(_annotation_equal(x, y) for x, y in zip(aa, ab))
setattr(result, k, copy(v))
return result
class IBaseElement(BaseModel, ABC): # ===================== Defaults mini-DSL (values only) =====================
_config_element: ClassVar[ConfigElement] = ConfigElement()
class _DefaultsSpec:
__slots__ = ("mapping",)
def __init__(self, mapping: Dict[str, Any]): self.mapping = mapping
@dataclass_transform(kw_only_default=True, field_specifiers=(PydanticModelField,)) def defaults(**kwargs) -> _DefaultsSpec:
class BaseElementMeta(ModelMetaclass, ABCMeta): """
def __new__( Declare class defaults in a flat mapping (values only).
mcs, Supports '.' or '__' for nesting:
cls_name: str, Defaults = defaults(
bases: tuple[type[Any], ...], template_short_name="apx",
namespace: dict[str, Any], Network__enabled=True,
__pydantic_generic_metadata__: PydanticGenericMetadata | None = None, Network__mtu=9000,
__pydantic_reset_parent_namespace__: bool = True, )
_create_model_module: str | None = None, """
**kwargs: Any, return _DefaultsSpec(kwargs)
) -> type:
result = super().__new__(
mcs,
cls_name,
bases,
namespace,
__pydantic_generic_metadata__,
__pydantic_reset_parent_namespace__,
_create_model_module,
**kwargs,
)
print(cls_name)
assert issubclass(result, IBaseElement), "Only IBaseElement subclasses are supported"
# forcing all Fields to be frozen def _split_path(key: str) -> list[str]:
for _, field_val in result.model_fields.items(): return key.replace("__", ".").split(".")
field_val.frozen = True
# copying/forwarding base classes default-configs
if "_config_element" not in result.__dict__: def _merge_path(target: Dict[str, Any], path: str, value: Any):
assert result.__base__ is not None, "Only IBaseElement subclasses are supported" parts = _split_path(path)
if issubclass(result.__base__, IBaseElement): node = target
result._config_element = copy(result.__base__._config_element) for i, p in enumerate(parts):
if i == len(parts) - 1:
node[p] = value
else:
node = node.setdefault(p, {}) # type: ignore[assignment]
def _flatten_defaults_class(cls: type) -> Dict[str, Any]:
out: Dict[str, Any] = {}
for name, val in vars(cls).items():
if name.startswith("__"):
continue
if isinstance(val, type):
sub = _flatten_defaults_class(val)
for k, v in sub.items():
out[f"{name}.{k}"] = v
else:
out[name] = val
return out
def _deep_merge(dst: Dict[str, Any], src: Dict[str, Any]):
for k, v in src.items():
if isinstance(v, dict) and isinstance(dst.get(k), dict):
_deep_merge(dst[k], v) # type: ignore[index]
else:
dst[k] = v
# ===================== Core base classes =====================
T_Feature = TypeVar("T_Feature", bound="BaseFeature")
class BaseElement(BaseModel):
"""
Base for Appliance/Feature:
- Declarative Defaults (inner class or mapping), values only
- Allowed-type enforcement at class creation (schema-time)
- Frozen instances
"""
model_config = ConfigDict(
frozen=True,
extra="forbid",
validate_assignment=True,
protected_namespaces=(), # allow names like __defaults_map__
)
__defaults_map__: Dict[str, Any] = {}
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
# 1) Collect Defaults (values) along MRO (base -> subclass)
merged: Dict[str, Any] = {}
for base in reversed(cls.mro()): # parent first, subclass last (subclass wins)
spec = base.__dict__.get("Defaults")
if spec is None:
continue
if isinstance(spec, _DefaultsSpec):
flat = spec.mapping
elif isinstance(spec, type):
flat = _flatten_defaults_class(spec)
else: else:
result._config_element = ConfigElement() continue
for k, v in flat.items():
_merge_path(merged, k, v)
# searching and storing current class default-configs # keep 'enabled' inside this map; we'll strip at validation time
for _, method in result.__dict__.items(): cls.__defaults_map__ = merged
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 # 2) Schema-time allowed types for non-feature fields
is_appl = any(b.__name__ == "BaseAppliance" for b in cls.mro())
is_feat = any(b.__name__ == "BaseFeature" for b in cls.mro())
if is_appl or is_feat:
allowed = ALLOWED_APPLIANCE_FIELD_TYPES if is_appl else ALLOWED_FEATURE_FIELD_TYPES
for name, f in getattr(cls, "model_fields", {}).items():
ann = f.annotation
if ann is None:
continue
if _is_feature_annotation(ann):
continue # feature fields checked in ApplianceMeta
if not _is_allowed_type(ann, allowed):
kind = "Appliance" if is_appl else "Feature"
raise TypeError(f"{kind} field '{cls.__name__}.{name}' has disallowed type: {ann}")
return result # >>> THIS MUST BE INSIDE THE CLASS <<<
@model_validator(mode="before")
@classmethod
def _apply_defaults_then_input(cls, values: Any):
if not isinstance(values, dict):
return values
# 1) seed from declared field defaults
merged: Dict[str, Any] = {}
for name, f in cls.model_fields.items():
if f.default is not None:
merged[name] = f.default
elif getattr(f, "default_factory", None) is not None:
merged[name] = f.default_factory() # type: ignore[misc]
# 2) merge class Defaults (values only), then 3) user values
_deep_merge(merged, cls.__defaults_map__)
_deep_merge(merged, values)
# 3) per-feature handling
for name, f in cls.model_fields.items():
ann = f.annotation
base = get_args(ann)[0] if get_origin(ann) is Optional else ann
try:
if isinstance(base, type) and issubclass(base, BaseFeature):
# user override flag?
user_block = values.get(name) if isinstance(values, dict) else None
user_enabled = None
if isinstance(user_block, dict) and "enabled" in user_block:
user_enabled = bool(user_block.get("enabled"))
# class default flag?
class_enabled = None
defaults_block = cls.__defaults_map__.get(name) if isinstance(cls.__defaults_map__, dict) else None
if isinstance(defaults_block, dict) and "enabled" in defaults_block:
class_enabled = bool(defaults_block.get("enabled"))
v = merged.get(name)
# *** fallback: read flag from merged payload if still present ***
merged_enabled = None
if isinstance(v, dict) and "enabled" in v:
merged_enabled = bool(v.get("enabled"))
# final precedence: user > class > merged
final_enabled = (
user_enabled
if user_enabled is not None
else (class_enabled if class_enabled is not None else merged_enabled)
)
# strip any lingering 'enabled' key so the feature never sees it
if isinstance(v, dict) and "enabled" in v:
v = dict(v)
v.pop("enabled", None)
merged[name] = v
if final_enabled is False:
merged[name] = None
continue
# Merge feature defaults to ensure required fields are present
feat_defaults = getattr(base, "__defaults_map__", {}) or {}
if final_enabled is True and v is None:
merged[name] = dict(feat_defaults)
elif isinstance(v, dict):
cfg: Dict[str, Any] = {}
_deep_merge(cfg, feat_defaults)
_deep_merge(cfg, v)
merged[name] = cfg
# If final_enabled is None and v is None → leave as None
except Exception:
pass
return merged
class BaseElement( class BaseFeature(BaseElement):
IBaseElement, template_id: UUID4
ABC, template_short_name: StrictStr
validate_assignment=True, template_long_name: Optional[StrictStr] = None
# revalidate_instances="subclass-instances", # pydantic issue #10681 template_description: Optional[StrictStr] = None
validate_default=True,
extra="forbid",
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") @model_validator(mode="before")
@classmethod @classmethod
def __default_values_override_hook__(cls, values: T_BaseElement_ConfigMethod_Arg) -> T_BaseElement_ConfigMethod_Arg: def _strip_enabled_flag(cls, v: Any):
# extracting default values that were set in model fields # Features never accept `enabled`; strip if present from Defaults or user input.
cls._saved_default_value = dict() if isinstance(v, dict) and "enabled" in v:
for field_key, field_val in cls.model_fields.items(): v = dict(v)
assert field_val.annotation is not None, "all fields must have annotation" v.pop("enabled", None)
assert not runtype_issubclass( return v
field_val.annotation, BaseFeature
), "Features can only be in Appliance's features[] dict attribute"
"""
if field_key == "features":
cls._saved_default_value[field_key] = dict()
for feat_key, feat_value in field_val.items():
cls._saved_default_value[field_key][feat_key] = feat_value.dict()
"""
if field_val.default != PydanticUndefined:
cls._saved_default_value[field_key] = deepcopy(field_val.default)
for method, _ in cls._config_element.default_values_override_methods.items():
method.__func__(cls, cls._saved_default_value)
cls._default_values_override_hook__input_apply__(values)
return cls._saved_default_value
@classmethod
@abstractmethod
def _default_values_override_hook__input_apply__(
cls,
values: T_BaseElement_ConfigMethod_Arg,
): ...
class BaseFeature(BaseElement, ABC): class ApplianceMeta(ModelMetaclass):
@NoInstanceMethod def __new__(mcls, name, bases, namespace, **kwargs):
@classmethod # Skip special handling for the abstract BaseAppliance itself
def _default_values_override_hook__input_apply__( if name == 'BaseAppliance':
cls, return super().__new__(mcls, name, bases, namespace, **kwargs)
values: T_BaseElement_ConfigMethod_Arg,
): # 1) Auto-inject feature fields from THIS class's inner `Features`
print(f"BaseFeature._default_values_override_hook__input_apply__ {cls}") declared_here: Dict[str, type[BaseFeature]] = {}
# applying user-defined values inner = namespace.get("Features")
for attr_key, attr_val in values.items(): if isinstance(inner, type):
assert attr_key in cls.model_fields, f"given feature attribute does not exist ({attr_key})" for fname, fval in vars(inner).items():
cls._saved_default_value[attr_key] = attr_val if fname.startswith("__"):
print(f"BaseFeature._default_values_override_hook__input_apply__ {cls} DONE") continue
if isinstance(fval, type) and issubclass(fval, BaseFeature):
declared_here[fname] = fval
namespace.setdefault("__annotations__", {})
if fname not in namespace["__annotations__"]:
namespace["__annotations__"][fname] = Optional[fval] # type: ignore[index]
namespace[fname] = None # default None; Defaults can enable/configure
# 2) Create the class via Pydantic's metaclass
cls = super().__new__(mcls, name, bases, namespace, **kwargs)
# 3) Discover current feature fields (after Pydantic processed annotations)
features_now: Dict[str, type[BaseFeature]] = {}
for field_name, f in cls.model_fields.items():
ann = f.annotation
base = get_args(ann)[0] if get_origin(ann) is Optional else ann
try:
if issubclass(base, BaseFeature): # type: ignore[arg-type]
features_now[field_name] = base # type: ignore[assignment]
except Exception:
pass
# 4) Enforce parent schema monotonicity (no removal/type changes), allow additions
parent = next((b for b in cls.__mro__[1:] if isinstance(b, ApplianceMeta)), None)
parent_fields = dict(getattr(parent, "model_fields", {})) if parent else {}
parent_features = getattr(parent, "__feature_fields__", {}) if parent else {}
# PARENT FIELDS: present with identical annotation + same constraints (skip feature fields)
for pname, pfield in parent_fields.items():
if pname in parent_features:
continue
if pname not in cls.model_fields:
raise TypeError(f"{cls.__name__}: removing parent field '{pname}' is forbidden.")
child_field = cls.model_fields[pname]
if (not _annotation_equal(child_field.annotation, pfield.annotation)) or (tuple(child_field.metadata) != tuple(pfield.metadata)):
raise TypeError(f"{cls.__name__}: changing type/constraints of parent field '{pname}' is forbidden.")
# PARENT FEATURES: present with identical type
for fname, ftype in parent_features.items():
if fname not in features_now:
raise TypeError(f"{cls.__name__}: removing parent feature '{fname}' is forbidden.")
if features_now[fname] is not ftype:
raise TypeError(f"{cls.__name__}: retargeting feature '{fname}' is forbidden.")
# NEW FEATURES must be declared in THIS class's inner Features
new_feature_names = set(features_now) - set(parent_features)
if new_feature_names:
if not declared_here:
raise TypeError(
f"{cls.__name__}: new features detected {sorted(new_feature_names)} "
f"but none declared in this class's inner `Features`."
)
undeclared = [n for n in new_feature_names if n not in declared_here]
if undeclared:
raise TypeError(
f"{cls.__name__}: new features {sorted(undeclared)} must be declared in this class's `Features`."
)
# 5) Publish the canonical sets
if parent and getattr(parent, "__declared_features__", {}):
merged_decl = dict(parent.__declared_features__)
merged_decl.update(declared_here)
cls.__declared_features__ = merged_decl
else:
cls.__declared_features__ = declared_here
cls.__feature_fields__ = features_now
return cls
def default_values_override(func: T_BaseElement_ConfigMethod) -> T_BaseElement_ConfigMethod: class BaseAppliance(BaseElement, Generic[T_Feature], metaclass=ApplianceMeta):
func = ensure_classmethod_based_on_signature(func) """
setattr(func, "default_values_override", lambda: True) Feature binding is implicit via inner `class Features:`.
return func Subclasses may add fields and add features freely.
They may NOT change types of parent fields/features or remove them.
"""
__declared_features__: Dict[str, type[BaseFeature]] = {}
__feature_fields__: Dict[str, type[BaseFeature]] = {}
# core appliance schema
template_id: UUID4
template_short_name: StrictStr
template_long_name: Optional[StrictStr] = None
template_description: Optional[StrictStr] = None
T_Feature = TypeVar("T_Feature", bound=BaseFeature) cpu_cnt: StrictInt = Field(1, gt=0)
ram_size: ByteSize = Field(256, gt=128)
swap_size: ByteSize = Field(200, ge=0)
rootfs_size: ByteSize = Field(2048, ge=2048)
dabinst_id: UUID4 = Field(default_factory=uuid4)
def get_discriminator_value(v: Any) -> str: dabinst_short_name: StrictStr
if isinstance(v, dict): dabinst_long_name: Optional[StrictStr] = ""
return v.get("fruit", v.get("filling")) dabinst_description: Optional[StrictStr] = ""
return getattr(v, "fruit", getattr(v, "filling", None)) dabinst_creationdate: Optional[AwareDatetime] = Field(default_factory=lambda: _dt.now(tz=pytz.utc))
class BaseAppliance(Generic[T_Feature], 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)]
rootfs_size: Annotated[ByteSize, DABField(2048, ge=2048)]
dabinst_id: Annotated[UUID4, DABField(uuid4(), repr=True)]
dabinst_short_name: Annotated[
StrictStr, constr(strip_whitespace=True, to_lower=True, strict=True, max_length=16), DABField(..., repr=True)
]
dabinst_long_name: Annotated[StrictStr | None, DABField("")]
dabinst_description: Annotated[StrictStr | None, DABField("")]
dabinst_creationdate: Annotated[AwareDatetime | None, DABField(datetime.now(tz=pytz.utc))]
features: SerializeAsAny[dict[str, T_Feature]] = DABField({})
@NoInstanceMethod
@classmethod
def add_feature(cls, feat: T_Feature):
cls._saved_default_value["features"][type(feat).__name__] = feat.dict()
@NoInstanceMethod
@classmethod
def del_feature(cls, type_feat: type[T_Feature]):
del cls._saved_default_value["features"][type_feat.__name__]
@NoInstanceMethod
@classmethod
def get_feature(cls, type_feat: type[T_Feature]) -> T_Feature:
return cls._saved_default_value["features"][type_feat.__name__]
@NoInstanceMethod
@classmethod
def _default_values_override_hook__input_apply__(
cls,
values: T_BaseElement_ConfigMethod_Arg,
):
print(f"BaseAppliance._default_values_override_hook__input_apply__ {cls}")
# applying user-defined values
for attr_key, attr_val in values.items():
if attr_key == "features":
if cls._saved_default_value["features"] is None:
cls._saved_default_value["features"] = {}
for feature_key, feature_val in attr_val.items():
print(f"searching feature: {feature_key}")
assert hasattr(cls, feature_key), f"feature not found ({feature_key})"
cls_feature = getattr(cls, feature_key)
assert (
cls_feature is not None and inspect.isclass(cls_feature) and issubclass(cls_feature, BaseFeature),
"The requested feature does not exist in the current Appliance class tree",
)
cls._saved_default_value["features"][feature_key] = cls_feature(**feature_val)
else:
assert attr_key in cls.model_fields, f"given attribute does not exist ({attr_key})"
cls._saved_default_value[attr_key] = attr_val
print(f"BaseAppliance._default_values_override_hook__input_apply__ {cls} DONE")

View File

@@ -1,37 +0,0 @@
from pydantic import BaseModel, SerializeAsAny
class commonbase(
BaseModel,
revalidate_instances="subclass-instances", # toogle to generate error
): ...
class basechild(commonbase):
test_val: int = 1
class derivedchild(basechild):
test_val2: int = 2
class container(commonbase):
ct_child_1: dict[str, basechild] = {}
ct_child_2: SerializeAsAny[dict[str, basechild]] = {}
ct_child_3: dict[str, SerializeAsAny[basechild]] = {}
if __name__ == "__main__":
test_val = container(
ct_child_1={"test1": derivedchild()},
ct_child_2={"test2": derivedchild()},
ct_child_3={"test3": derivedchild()},
)
print(test_val.model_dump_json(indent=1))
print(test_val.model_dump())
assert "test_val2" not in test_val.model_dump()["ct_child_1"]["test1"]
assert "test_val2" in test_val.model_dump()["ct_child_2"]["test2"]
assert "test_val2" in test_val.model_dump()["ct_child_3"]["test3"]

View File

@@ -10,14 +10,13 @@ import unittest
from os import chdir from os import chdir
from pathlib import Path from pathlib import Path
from pydantic import StrictInt, model_validator
from pydantic.fields import Field
print(__name__) print(__name__)
print(__package__) print(__package__)
from src import dabmodel from src import dabmodel as dm
from typing import Annotated, Any from typing import Annotated, Any, Optional
from uuid import uuid4 from uuid import uuid4
import json import json
from uuid import UUID from uuid import UUID
@@ -26,7 +25,7 @@ from datetime import datetime
testdir_path = Path(__file__).parent.resolve() testdir_path = Path(__file__).parent.resolve()
chdir(testdir_path.parent.resolve()) chdir(testdir_path.parent.resolve())
"""
class UUIDEncoder(json.JSONEncoder): class UUIDEncoder(json.JSONEncoder):
def default(self, obj): def default(self, obj):
if isinstance(obj, UUID): if isinstance(obj, UUID):
@@ -35,187 +34,137 @@ class UUIDEncoder(json.JSONEncoder):
elif isinstance(obj, datetime): elif isinstance(obj, datetime):
return str(obj) return str(obj)
return json.JSONEncoder.default(self, obj) return json.JSONEncoder.default(self, obj)
"""
from typing import Optional
from pydantic import ValidationError,StrictInt,StrictStr
class MyAppliance(dabmodel.BaseAppliance):
app_specifi_integer_arg: Annotated[StrictInt | None, dabmodel.DABField(42)]
class MyFeature(dabmodel.BaseFeature): # ---------------- Basic root appliance with one feature ----------------
@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): class Router(dm.BaseAppliance):
@dabmodel.default_values_override class Features:
@classmethod class Network(dm.BaseFeature):
def __override_config__(cls, values): mtu: StrictInt = 1500
print("!!! CONFIG Feature 2") class Defaults:
values["template_id"] = "421d61cb-e364-46d8-9b77-ec439f1fa666" template_id = "00000000-0000-4000-8000-000000000001"
values["template_short_name"] = "my-feature-2" template_short_name = "net"
values["template_long_name"] = "My feature template 2 !!"
values["template_description"] = """A very nice FEature 2"""
@dabmodel.default_values_override class Defaults:
@classmethod template_id = "11111111-1111-4111-8111-111111111111"
def __override_config__(cls, values): template_short_name = "router"
print("!!! CONFIG Appliance 1") class Network:
print(f"!!!! {values['rootfs_size']}") enabled = True
values["template_id"] = "421d61cb-e364-46d8-9b64-ec439f1faae8" mtu = 9000
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
cls.add_feature(cls.MyFeature())
cls.add_feature(cls.MyFeature2())
class MyAppliance2(MyAppliance): # ---------------- Subclass adds fields and features (allowed) ----------------
@dabmodel.default_values_override
@classmethod class RouterPlus(Router):
def __override_config__(cls, values): model_name: StrictStr = "plus" # new non-feature field (allowed)
print("!!! CONFIG Appliance 2")
print(f"!!!! {values['template_id']}") class Features:
values["template_id"] = "421d61cb-e664-46d8-9b64-ec439f1fafff" class Firewall(dm.BaseFeature):
values["template_short_name"] = "my-app- tem 2" enabled_by_default: bool = True
values["template_long_name"] = "My appliance template 2 !!" class Defaults:
values["template_description"] = """A very nice Appliance 2""" template_id = "22222222-2222-4222-8222-222222222222"
cls.del_feature(MyAppliance.MyFeature) template_short_name = "fw"
# values["features"]["MyFeature2"].template_description = """Override feature desc"""
class Defaults:
class Firewall:
enabled = True
enabled_by_default = True
class MyAppliance3(dabmodel.BaseAppliance): # ---------------- Subclass tweaks values only ----------------
class MyFeature6(MyAppliance.MyFeature): class RouterLite(RouterPlus):
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case class Defaults:
test_integer: Annotated[int, dabmodel.DABField(200, ge=0)] template_short_name = "router-lite"
model_name = "lite"
@dabmodel.default_values_override class Network:
@classmethod enabled = False # allowed: disable inherited feature
def __override_config__(cls, values): class Firewall:
print("!!! CONFIG Feature 1 (modified)") enabled = True
values["template_id"] = "421d61cb-e364-46d8-9b77-ec439f1fa778" enabled_by_default = False
values["template_short_name"] = "my-feature-1-bis"
values["test_integer"] = 666
class MyFeature7(dabmodel.BaseFeature):
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case
test_integer_2: Annotated[int, dabmodel.DABField(759, ge=0)]
@dabmodel.default_values_override
@classmethod
def __override_config__(cls, values):
print("!!! CONFIG Feature 7")
values["template_id"] = "421d61cb-e364-46d8-ac55-ec439f1fa778"
values["template_short_name"] = "my-feature-7"
values["template_long_name"] = "My appliance template 7 !!"
values["template_description"] = """A very nice Appliance 7"""
values["test_integer_2"] = 3844
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case
@dabmodel.default_values_override
@classmethod
def __override_config__(cls, values):
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
print("CREATE FEATURE")
cls.add_feature(cls.MyFeature6())
cls.add_feature(cls.MyFeature7())
print("!!! CONFIG Appliance 3 DONE")
class MyAppliance4(MyAppliance): # ---------------- Negative: remove parent feature ----------------
class MyFeature8(dabmodel.BaseFeature): def _make_bad_remove_feature():
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case (nested feature) class Bad(RouterPlus):
# test_integer_10: Annotated[int, dabmodel.DABField(3189, ge=0, toto="tata")] # error case (extra field) # Attempt to remove Network by redefining inner Features without it
test_integer_10: Annotated[int, dabmodel.DABField(3189, ge=0, toto="tata")] class Features:
class Firewall(dm.BaseFeature):
@dabmodel.default_values_override enabled_by_default: bool = True
@classmethod pass
def __override_config__(cls, values): return Bad
print("!!! CONFIG Feature 8")
values["template_id"] = "421d61cb-e364-46d8-ac55-ec4398888778"
values["template_short_name"] = "my-feature-8"
values["template_long_name"] = "My appliance template 8 !!"
values["template_description"] = """A very nice Appliance 8"""
values["test_integer_10"] = 951753
# values["tete"] = 1 # error case (extra field in feature)
# testtt: Annotated[MyAppliance.MyFeature, Field(MyAppliance.MyFeature())] # error case (feature not in features[] list)
@dabmodel.default_values_override
@classmethod
def __override_config__(cls, values):
print("!!! CONFIG Appliance 4")
values["template_id"] = "421d1234-e364-46d8-9b64-ec439f1faaaa"
values["template_short_name"] = "my-app-tem 4"
values["template_long_name"] = "My appliance template 4 !!"
values["template_description"] = """A very nice Appliance 4"""
values["ram_size"] = 954
print("CREATE FEATURE")
cls.add_feature(cls.MyFeature8())
print("!!! CONFIG Appliance 4 DONE")
class TestModel(unittest.TestCase): # ---------------- Negative: change type of parent field ----------------
def setUp(self) -> None:
chdir(testdir_path.parent.resolve())
def test_version(self): def _make_bad_change_type():
self.assertNotEqual(dabmodel.__version__, "?.?.?") class Bad(Router):
cpu_cnt: int = 2 # lose StrictInt (type change forbidden)
return Bad
def test_model(self):
feature1 = MyAppliance.MyFeature() # ---------------- Negative: add new feature without declaring in Features ----------------
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) def _make_bad_undeclared_feature():
app2 = MyAppliance2(dabinst_short_name="my-app-2", app_specifi_integer_arg=654) class Bad(Router):
app3 = MyAppliance3(dabinst_short_name="my-app-3", template_description="FORCED") # Inject an extra feature field not present in this class's `Features`
class Features:
class Network(dm.BaseFeature):
mtu: StrictInt = 1500
class Extra(dm.BaseFeature):
foo: StrictInt = 1
Extra: Optional[Extra] = Extra()
return Bad
print(app.model_dump_json(indent=1))
print(app2.model_dump_json(indent=1))
print(app3.model_dump_json(indent=1))
app3 = MyAppliance3(dabinst_short_name="my-app-3", template_description="FORCED") # ---------------- Tests ----------------
tmp_json = app3.dict()
tmp_json["features"]["MyFeature7"]["test_integer_2"] = 123
print(tmp_json)
recreated_obj = MyAppliance3.model_validate_json(json.dumps(tmp_json, cls=UUIDEncoder))
print(recreated_obj)
print(recreated_obj.model_dump_json(indent=1))
app4 = MyAppliance4(dabinst_short_name="my-app-4", template_description="FORCED2") class TestDabModelV2(unittest.TestCase):
tmp_json = app4.dict() def test_root_defaults_and_feature(self):
tmp_json["features"]["MyFeature"]["template_description"] = "blablabla" a = Router(dabinst_short_name="r1")
tmp_json["features"]["MyFeature2"]["template_description"] = "blablabla2" self.assertEqual(a.template_short_name, "router")
print(tmp_json) self.assertIsNotNone(a.Network)
recreated_obj = MyAppliance4.model_validate_json(json.dumps(tmp_json, cls=UUIDEncoder)) self.assertEqual(a.Network.mtu, 9000)
print(recreated_obj)
print(recreated_obj.model_dump_json(indent=1))
# tmp_json["non-existing"] = "test" # error case def test_subclass_adds_field_and_feature(self):
# tmp_json["features"]["non-existing"] = "test" # error case a = RouterPlus(dabinst_short_name="rp1")
# tmp_json["features"]["MyFeature"]["132"] = "test" # error case self.assertEqual(a.model_name, "plus")
self.assertIsNotNone(a.Network)
self.assertIsNotNone(a.Firewall)
self.assertTrue(a.Firewall.enabled_by_default)
recreated_obj = MyAppliance4.model_validate_json(json.dumps(tmp_json, cls=UUIDEncoder)) def test_subclass_value_overrides(self):
a = RouterLite(dabinst_short_name="rl1")
self.assertEqual(a.template_short_name, "router-lite")
self.assertEqual(a.model_name, "lite")
self.assertIsNone(a.Network) # disabled
self.assertIsNotNone(a.Firewall)
self.assertFalse(a.Firewall.enabled_by_default)
def test_user_input_overrides(self):
a = Router(dabinst_short_name="u1", Network={'mtu': 4096})
self.assertEqual(a.Network.mtu, 4096)
b = RouterPlus(dabinst_short_name="u2", Firewall={'enabled': False})
self.assertIsNone(b.Firewall)
def test_cannot_remove_parent_feature(self):
with self.assertRaises(TypeError):
_ = _make_bad_remove_feature()
def test_cannot_change_parent_field_type(self):
with self.assertRaises(TypeError):
_ = _make_bad_change_type()
def test_new_feature_must_be_declared_in_features(self):
with self.assertRaises(ValidationError):
_ = _make_bad_undeclared_feature()
# app3.add_feature(MyAppliance.MyFeature()) # error case (add_feature not callable from instance)
for name in globals().keys():
print(name)