Compare commits
4 Commits
0.0.1.post
...
dev
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
85a2ca2753 | ||
|
|
df50632458 | ||
|
|
25d5339946 | ||
|
|
6efd914de1 |
@@ -2,14 +2,14 @@ from typing import Any, Union, Optional, List, Dict, Tuple, Set, FrozenSet, Anno
|
||||
from types import SimpleNamespace
|
||||
import math
|
||||
|
||||
ALLOWED_MODEL_FIELDS_TYPES: tuple[type[Any], ...] = (
|
||||
ALLOWED_MODEL_FIELDS_TYPES: set[type[Any], ...] = {
|
||||
str,
|
||||
int,
|
||||
float,
|
||||
complex,
|
||||
bool,
|
||||
bytes,
|
||||
)
|
||||
}
|
||||
|
||||
ALLOWED_ANNOTATIONS: dict[str, Any] = {
|
||||
"Union": Union,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from typing import Generic, TypeVar, Optional, Any, Self
|
||||
from typing import Generic, TypeVar, Optional, Any, Self, Annotated, get_origin, get_args
|
||||
from typeguard import check_type, CollectionCheckStrategy, TypeCheckError
|
||||
from copy import deepcopy
|
||||
from .lam_field_info import LAMFieldInfo
|
||||
@@ -13,19 +13,29 @@ TV_LABField = TypeVar("TV_LABField")
|
||||
class LAMField(Generic[TV_LABField]):
|
||||
"""This class describe a Field in Schema"""
|
||||
|
||||
def __init__(self, name: str, val: Optional[TV_LABField], a: Any, i: LAMFieldInfo):
|
||||
def __init__(self, name: str, val: Optional[TV_LABField], ann: Any, i: LAMFieldInfo):
|
||||
self._default_value: Optional[TV_LABField]
|
||||
self._value: Optional[TV_LABField]
|
||||
self.__annotations: Any
|
||||
self.__name: str = name
|
||||
self.__source: Optional[type] = None
|
||||
self.__info: LAMFieldInfo = deepcopy(i)
|
||||
self.__annotations: Any = LAMdeepfreeze(a)
|
||||
self._set_annotations(ann)
|
||||
self.__frozen: bool = False
|
||||
self._frozen_value: Any = None
|
||||
self.__frozen_value_set: True = False
|
||||
self.validate(val)
|
||||
self._init_value(val)
|
||||
|
||||
def _set_annotations(self, ann: Any) -> None:
|
||||
_origin = get_origin(ann) or ann
|
||||
_args = get_args(ann)
|
||||
|
||||
if _origin is Annotated:
|
||||
self.__annotations: Any = LAMdeepfreeze(_args[0])
|
||||
else:
|
||||
self.__annotations: Any = LAMdeepfreeze(ann)
|
||||
|
||||
def _init_value(self, val: Optional[TV_LABField | FreezableElement]):
|
||||
self._default_value: Optional[TV_LABField] = deepcopy(val)
|
||||
self._value: Optional[TV_LABField] = val
|
||||
|
||||
@@ -177,7 +177,7 @@ class IAppliance(IBaseElement): ...
|
||||
|
||||
|
||||
def _check_annotation_definition(_type, first_layer: bool = True): # pylint: disable=too-complex,too-many-return-statements
|
||||
# print(f"_type={_type}, {first_layer}")
|
||||
print(f"_type={_type}, {first_layer}")
|
||||
|
||||
_origin = get_origin(_type) or _type
|
||||
_args = get_args(_type)
|
||||
@@ -194,7 +194,7 @@ def _check_annotation_definition(_type, first_layer: bool = True): # pylint: di
|
||||
raise UnsupportedFieldType(
|
||||
"Union[] is only supported to implement Optional (takes 2 parameters) and is only supported as parent annotation"
|
||||
)
|
||||
return all(_check_annotation_definition(_, False) for _ in get_args(_type) if _ is not type(None))
|
||||
return any(_check_annotation_definition(_, False) for _ in get_args(_type) if _ is not type(None))
|
||||
|
||||
# handle Dict[...]
|
||||
if _origin is dict:
|
||||
@@ -202,7 +202,8 @@ def _check_annotation_definition(_type, first_layer: bool = True): # pylint: di
|
||||
raise IncompletelyAnnotatedField(f"Dict Annotation requires 2 inner definitions: {_type}")
|
||||
if not _args[0] in ALLOWED_MODEL_FIELDS_TYPES:
|
||||
raise IncompletelyAnnotatedField(f"Dict Key must be simple builtin: {_type}")
|
||||
return _check_annotation_definition(_args[1], False)
|
||||
# return _check_annotation_definition(_args[1], False)
|
||||
return any(_check_annotation_definition(_, False) for _ in _args)
|
||||
|
||||
# handle Tuple[]
|
||||
if _origin is tuple:
|
||||
@@ -210,13 +211,13 @@ def _check_annotation_definition(_type, first_layer: bool = True): # pylint: di
|
||||
raise IncompletelyAnnotatedField(f"Annotation requires inner definition: {_type}")
|
||||
if len(_args) == 2 and _args[1] is Ellipsis:
|
||||
return _check_annotation_definition(_args[0], False)
|
||||
return all(_check_annotation_definition(_, False) for _ in _args)
|
||||
return any(_check_annotation_definition(_, False) for _ in _args)
|
||||
|
||||
# handle Set[],Tuple[],FrozenSet[],List[]
|
||||
if _origin in [set, frozenset, tuple, list]:
|
||||
if len(_args) == 0:
|
||||
raise IncompletelyAnnotatedField(f"Annotation requires inner definition: {_type}")
|
||||
return all(_check_annotation_definition(_, False) for _ in _args)
|
||||
return any(_check_annotation_definition(_, False) for _ in _args)
|
||||
|
||||
if isinstance(_origin, type):
|
||||
if issubclass(_origin, IElement):
|
||||
@@ -224,9 +225,9 @@ def _check_annotation_definition(_type, first_layer: bool = True): # pylint: di
|
||||
elif issubclass(_origin, IAppliance):
|
||||
raise UnsupportedFieldType(f"Nested Appliance are not supported: {_type}")
|
||||
|
||||
if _type in ALLOWED_MODEL_FIELDS_TYPES:
|
||||
if _origin in ALLOWED_MODEL_FIELDS_TYPES:
|
||||
return
|
||||
raise UnsupportedFieldType(_type)
|
||||
raise UnsupportedFieldType(_origin)
|
||||
|
||||
|
||||
def _check_initializer_safety(func) -> None:
|
||||
|
||||
@@ -1,16 +1,36 @@
|
||||
"""library's internal tools"""
|
||||
|
||||
from typing import Any, get_origin, get_args
|
||||
from collections import ChainMap
|
||||
from typing import (
|
||||
Any,
|
||||
Annotated,
|
||||
get_origin,
|
||||
get_args,
|
||||
Union,
|
||||
Self,
|
||||
Optional,
|
||||
List,
|
||||
Dict,
|
||||
Tuple,
|
||||
Set,
|
||||
FrozenSet,
|
||||
Mapping,
|
||||
Callable,
|
||||
)
|
||||
import typing
|
||||
from dataclasses import dataclass
|
||||
from types import UnionType, NoneType
|
||||
import types
|
||||
from uuid import UUID
|
||||
from datetime import datetime
|
||||
import json
|
||||
import inspect
|
||||
|
||||
from frozendict import deepfreeze
|
||||
|
||||
from .defines import (
|
||||
ALLOWED_ANNOTATIONS,
|
||||
)
|
||||
from frozendict import deepfreeze, frozendict
|
||||
|
||||
from .defines import ALLOWED_ANNOTATIONS, ALLOWED_MODEL_FIELDS_TYPES
|
||||
from .exception import IncompletelyAnnotatedField, UnsupportedFieldType
|
||||
|
||||
|
||||
class LAMJSONEncoder(json.JSONEncoder):
|
||||
@@ -48,22 +68,6 @@ def is_data_attribute(name: str, value: any) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
"""
|
||||
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 _resolve_annotation(ann):
|
||||
if isinstance(ann, str):
|
||||
# Safe eval against a **whitelist** only
|
||||
|
||||
53
test/test_AnnotationTool.py
Normal file
53
test/test_AnnotationTool.py
Normal file
@@ -0,0 +1,53 @@
|
||||
# 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 typing import Annotated, Union, Optional
|
||||
import sys
|
||||
import subprocess
|
||||
from os import chdir, environ
|
||||
from pathlib import Path
|
||||
|
||||
print(__name__)
|
||||
print(__package__)
|
||||
|
||||
from src import dabmodel as dm
|
||||
|
||||
testdir_path = Path(__file__).parent.resolve()
|
||||
chdir(testdir_path.parent.resolve())
|
||||
|
||||
|
||||
class AnnotationsWalkerTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
print("\n->", unittest.TestCase.id(self))
|
||||
|
||||
def test_validate(self):
|
||||
ann = dict[int, list[int]] | dict[int, list[int | str]]
|
||||
val = {1: [2], 2: ["a", [1]]}
|
||||
|
||||
res = dm.tools.AnnotationWalker(ann, (dm.tools.HorizontalValidationTrigger(val),))
|
||||
res.run()
|
||||
|
||||
def test_simple(self):
|
||||
print(isinstance(None, type(None)))
|
||||
print("\n== From OBJs ==")
|
||||
res = dm.tools.AnnotationWalker(Annotated[Optional[dict[int, list[str]]], "comment"], (dm.tools.LAMSchemaValidation(),))
|
||||
print(f"res={res.run()}")
|
||||
|
||||
print("\n== From STRING ==")
|
||||
res = dm.tools.AnnotationWalker('Annotated[Optional[dict[int, list[str]]], "comment"]', (dm.tools.LAMSchemaValidation(),))
|
||||
print(f"res={res.run()}")
|
||||
|
||||
res = dm.tools.AnnotationWalker(Annotated[Optional[dict[int, list[None]]], "comment"], (dm.tools.LAMSchemaValidation(),))
|
||||
print(f"res={res.run()}")
|
||||
|
||||
|
||||
# ---------- main ----------
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user