from knot_resolver_manager.utils.functional import all_matches
+from .base_generic_type_wrapper import BaseGenericTypeWrapper
from .base_value_type import BaseValueType
from .exceptions import AggregateDataValidationError, DataDescriptionError, DataValidationError
from .renaming import Renamed, renamed
from .types import (
get_generic_type_argument,
get_generic_type_arguments,
+ get_generic_type_wrapper_argument,
get_optional_inner_type,
is_dict,
is_enum,
+ is_generic_type_wrapper,
is_internal_field_name,
is_list,
is_literal,
or is_literal(typ)
or is_dict(typ)
or is_list(typ)
+ or is_generic_type_wrapper(typ)
or (inspect.isclass(typ) and issubclass(typ, Serializable))
or (inspect.isclass(typ) and issubclass(typ, BaseValueType))
or (inspect.isclass(typ) and issubclass(typ, BaseSchema))
if isinstance(obj, Serializable):
return obj.to_dict()
- elif isinstance(obj, BaseValueType):
- return obj.serialize()
+ elif isinstance(obj, (BaseValueType, BaseGenericTypeWrapper)):
+ o = obj.serialize()
+ # if Serializable.is_serializable(o):
+ return Serializable.serialize(o)
+ # return o
elif isinstance(obj, list):
res: List[Any] = [Serializable.serialize(i) for i in cast(List[Any], obj)]
elif inspect.isclass(typ) and issubclass(typ, BaseValueType):
return typ.json_schema()
+ elif is_generic_type_wrapper(typ):
+ wrapped = get_generic_type_wrapper_argument(typ)
+ return _describe_type(wrapped)
+
elif is_none_type(typ):
return {"type": "null"}
inner_type = get_generic_type_argument(tp)
errs: List[DataValidationError] = []
res: List[Any] = []
- for i, val in enumerate(obj):
- try:
+
+ try:
+ for i, val in enumerate(obj):
res.append(self.map_object(inner_type, val, object_path=f"{object_path}[{i}]"))
- except DataValidationError as e:
- errs.append(e)
+ except DataValidationError as e:
+ errs.append(e)
+ except TypeError as e:
+ errs.append(DataValidationError(str(e), object_path))
+
if len(errs) == 1:
raise errs[0]
elif len(errs) > 1:
elif inspect.isclass(tp) and issubclass(tp, BaseValueType):
return self.create_value_type_object(tp, obj, object_path)
+ # BaseGenericTypeWrapper subclasses
+ elif is_generic_type_wrapper(tp):
+ inner_type = get_generic_type_wrapper_argument(tp)
+ obj_valid = self.map_object(inner_type, obj, object_path)
+ return tp(obj_valid, object_path=object_path) # type: ignore
+
# nested BaseSchema subclasses
elif inspect.isclass(tp) and issubclass(tp, BaseSchema):
return self._create_base_schema_object(tp, obj, object_path)
from typing import Any, Dict, Type
-class BaseValueType(ABC):
- """
- Subclasses of this class can be used as type annotations in 'DataParser'. When a value
- is being parsed from a serialized format (e.g. JSON/YAML), an object will be created by
- calling the constructor of the appropriate type on the field value. The only limitation
- is that the value MUST NOT be `None`.
-
- There is no validation done on the wrapped value. The only condition is that
- it can't be `None`. If you want to perform any validation during creation,
- raise a `ValueError` in case of errors.
- """
-
+class BaseTypeABC(ABC):
@abstractmethod
def __init__(self, source_value: Any, object_path: str = "/") -> None:
pass
"""
raise NotImplementedError(f"{type(self).__name__}'s' 'serialize()' not implemented.")
+
+class BaseValueType(BaseTypeABC):
+ """
+ Subclasses of this class can be used as type annotations in 'DataParser'. When a value
+ is being parsed from a serialized format (e.g. JSON/YAML), an object will be created by
+ calling the constructor of the appropriate type on the field value. The only limitation
+ is that the value MUST NOT be `None`.
+
+ There is no validation done on the wrapped value. The only condition is that
+ it can't be `None`. If you want to perform any validation during creation,
+ raise a `ValueError` in case of errors.
+ """
+
@classmethod
@abstractmethod
def json_schema(cls: Type["BaseValueType"]) -> Dict[Any, Any]:
from typing_extensions import Literal
+from .base_generic_type_wrapper import BaseGenericTypeWrapper
+
NoneType = type(None)
return getattr(tp, "__origin__", None) == Literal
+def is_generic_type_wrapper(tp: Any) -> bool:
+ orig = getattr(tp, "__origin__", None)
+ return inspect.isclass(orig) and issubclass(orig, BaseGenericTypeWrapper)
+
+
def get_generic_type_arguments(tp: Any) -> List[Any]:
default: List[Any] = []
if sys.version_info.minor == 6 and is_literal(tp):
return args[0]
+def get_generic_type_wrapper_argument(tp: Type["BaseGenericTypeWrapper[Any]"]) -> Any:
+ assert hasattr(tp, "__origin__")
+ origin = getattr(tp, "__origin__")
+
+ assert hasattr(origin, "__orig_bases__")
+ orig_base: List[Any] = getattr(origin, "__orig_bases__", [])[0]
+
+ arg = get_generic_type_argument(tp)
+ return get_generic_type_argument(orig_base[arg])
+
+
def is_none_type(tp: Any) -> bool:
return tp is None or tp == NoneType
projects / thirdparty / knot-resolver.git / commitdiff
