1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820
|
import enum
import inspect
from abc import ABC, abstractmethod # pylint: disable=[no-name-in-module]
from typing import Any, Callable, Dict, Generic, List, Optional, Set, Tuple, Type, TypeVar, Union, cast
import yaml
from knot_resolver.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,
is_none_type,
is_optional,
is_tuple,
is_union,
)
T = TypeVar("T")
def is_obj_type(obj: Any, types: Union[type, Tuple[Any, ...], Tuple[type, ...]]) -> bool:
# To check specific type we are using 'type()' instead of 'isinstance()'
# because for example 'bool' is instance of 'int', 'isinstance(False, int)' returns True.
# pylint: disable=unidiomatic-typecheck
if isinstance(types, tuple):
return type(obj) in types
return type(obj) is types
class Serializable(ABC):
"""
An interface for making classes serializable to a dictionary (and in turn into a JSON).
"""
@abstractmethod
def to_dict(self) -> Dict[Any, Any]:
raise NotImplementedError(f"...for class {self.__class__.__name__}")
@staticmethod
def is_serializable(typ: Type[Any]) -> bool:
return (
typ in {str, bool, int, float}
or is_none_type(typ)
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))
or (is_optional(typ) and Serializable.is_serializable(get_optional_inner_type(typ)))
or (is_union(typ) and all_matches(Serializable.is_serializable, get_generic_type_arguments(typ)))
)
@staticmethod
def serialize(obj: Any) -> Any:
if isinstance(obj, Serializable):
return obj.to_dict()
if isinstance(obj, (BaseValueType, BaseGenericTypeWrapper)):
o = obj.serialize()
# if Serializable.is_serializable(o):
return Serializable.serialize(o)
# return o
if isinstance(obj, list):
res: List[Any] = [Serializable.serialize(i) for i in cast(List[Any], obj)]
return res
return obj
class _LazyDefault(Generic[T], Serializable):
"""
Wrapper for default values BaseSchema classes which deffers their instantiation until the schema
itself is being instantiated
"""
def __init__(self, constructor: Callable[..., T], *args: Any, **kwargs: Any) -> None:
# pylint: disable=[super-init-not-called]
self._func = constructor
self._args = args
self._kwargs = kwargs
def instantiate(self) -> T:
return self._func(*self._args, **self._kwargs)
def to_dict(self) -> Dict[Any, Any]:
return Serializable.serialize(self.instantiate())
def lazy_default(constructor: Callable[..., T], *args: Any, **kwargs: Any) -> T:
"""We use a factory function because you can't lie about the return type in `__new__`"""
return _LazyDefault(constructor, *args, **kwargs) # type: ignore
def _split_docstring(docstring: str) -> Tuple[str, Optional[str]]:
"""
Splits docstring into description of the class and description of attributes
"""
if "---" not in docstring:
return ("\n".join([s.strip() for s in docstring.splitlines()]).strip(), None)
doc, attrs_doc = docstring.split("---", maxsplit=1)
return (
"\n".join([s.strip() for s in doc.splitlines()]).strip(),
attrs_doc,
)
def _parse_attrs_docstrings(docstring: str) -> Optional[Dict[str, str]]:
"""
Given a docstring of a BaseSchema, return a dict with descriptions of individual attributes.
"""
_, attrs_doc = _split_docstring(docstring)
if attrs_doc is None:
return None
# try to parse it as yaml:
data = yaml.safe_load(attrs_doc)
assert isinstance(data, dict), "Invalid format of attribute description"
return cast(Dict[str, str], data)
def _get_properties_schema(typ: Type[Any]) -> Dict[Any, Any]:
schema: Dict[Any, Any] = {}
annot: Dict[str, Any] = typ.__dict__.get("__annotations__", {})
docstring: str = typ.__dict__.get("__doc__", "") or ""
attribute_documentation = _parse_attrs_docstrings(docstring)
for field_name, python_type in annot.items():
name = field_name.replace("_", "-")
schema[name] = _describe_type(python_type)
# description
if attribute_documentation is not None:
if field_name not in attribute_documentation:
raise DataDescriptionError(f"The docstring does not describe field '{field_name}'", str(typ))
schema[name]["description"] = attribute_documentation[field_name]
del attribute_documentation[field_name]
# default value
if hasattr(typ, field_name):
assert Serializable.is_serializable(
python_type
), f"Type '{python_type}' does not appear to be JSON serializable"
schema[name]["default"] = Serializable.serialize(getattr(typ, field_name))
if attribute_documentation is not None and len(attribute_documentation) > 0:
raise DataDescriptionError(
f"The docstring describes attributes which are not present - {tuple(attribute_documentation.keys())}",
str(typ),
)
return schema
def _describe_type(typ: Type[Any]) -> Dict[Any, Any]: # noqa: PLR0911, PLR0912
# pylint: disable=too-many-branches
if inspect.isclass(typ) and issubclass(typ, BaseSchema):
return typ.json_schema(include_schema_definition=False)
if inspect.isclass(typ) and issubclass(typ, BaseValueType):
return typ.json_schema()
if is_generic_type_wrapper(typ):
wrapped = get_generic_type_wrapper_argument(typ)
return _describe_type(wrapped)
if is_none_type(typ):
return {"type": "null"}
if typ is int:
return {"type": "integer"}
if typ is bool:
return {"type": "boolean"}
if typ is str:
return {"type": "string"}
if is_literal(typ):
lit = get_generic_type_arguments(typ)
return {"type": "string", "enum": lit}
if is_optional(typ):
desc = _describe_type(get_optional_inner_type(typ))
if "type" in desc:
desc["type"] = [desc["type"], "null"]
return desc
return {"anyOf": [{"type": "null"}, desc]}
if is_union(typ):
variants = get_generic_type_arguments(typ)
return {"anyOf": [_describe_type(v) for v in variants]}
if is_list(typ):
return {"type": "array", "items": _describe_type(get_generic_type_argument(typ))}
if is_dict(typ):
key, val = get_generic_type_arguments(typ)
if inspect.isclass(key) and issubclass(key, BaseValueType):
assert (
key.__str__ is not BaseValueType.__str__
), "To support derived 'BaseValueType', __str__ must be implemented."
else:
assert key is str, "We currently do not support any other keys then strings"
return {"type": "object", "additionalProperties": _describe_type(val)}
if inspect.isclass(typ) and issubclass(typ, enum.Enum): # same as our is_enum(typ), but inlined for type checker
return {"type": "string", "enum": [str(v) for v in typ]}
raise NotImplementedError(f"Trying to get JSON schema for type '{typ}', which is not implemented")
TSource = Union[None, "BaseSchema", Dict[str, Any]]
def _create_untouchable(name: str) -> object:
class _Untouchable:
def __getattribute__(self, item_name: str) -> Any:
raise RuntimeError(f"You are not supposed to access object '{name}'.")
def __setattr__(self, item_name: str, value: Any) -> None:
raise RuntimeError(f"You are not supposed to access object '{name}'.")
return _Untouchable()
class ObjectMapper:
def _create_tuple(self, tp: Type[Any], obj: Tuple[Any, ...], object_path: str) -> Tuple[Any, ...]:
types = get_generic_type_arguments(tp)
errs: List[DataValidationError] = []
res: List[Any] = []
for i, (t, val) in enumerate(zip(types, obj)):
try:
res.append(self.map_object(t, val, object_path=f"{object_path}[{i}]"))
except DataValidationError as e:
errs.append(e)
if len(errs) == 1:
raise errs[0]
if len(errs) > 1:
raise AggregateDataValidationError(object_path, child_exceptions=errs)
return tuple(res)
def _create_dict(self, tp: Type[Any], obj: Dict[Any, Any], object_path: str) -> Dict[Any, Any]:
key_type, val_type = get_generic_type_arguments(tp)
try:
errs: List[DataValidationError] = []
res: Dict[Any, Any] = {}
for key, val in obj.items():
try:
nkey = self.map_object(key_type, key, object_path=f"{object_path}[{key}]")
nval = self.map_object(val_type, val, object_path=f"{object_path}[{key}]")
res[nkey] = nval
except DataValidationError as e:
errs.append(e)
if len(errs) == 1:
raise errs[0]
if len(errs) > 1:
raise AggregateDataValidationError(object_path, child_exceptions=errs)
return res
except AttributeError as e:
raise DataValidationError(
f"Expected dict-like object, but failed to access its .items() method. Value was {obj}", object_path
) from e
def _create_list(self, tp: Type[Any], obj: List[Any], object_path: str) -> List[Any]:
if isinstance(obj, str):
raise DataValidationError("expected list, got string", object_path)
inner_type = get_generic_type_argument(tp)
errs: List[DataValidationError] = []
res: List[Any] = []
try:
for i, val in enumerate(obj):
res.append(self.map_object(inner_type, val, object_path=f"{object_path}[{i}]"))
if len(res) == 0:
raise DataValidationError("empty list is not allowed", object_path)
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]
if len(errs) > 1:
raise AggregateDataValidationError(object_path, child_exceptions=errs)
return res
def _create_str(self, obj: Any, object_path: str) -> str:
# we are willing to cast any primitive value to string, but no compound values are allowed
if is_obj_type(obj, (str, float, int)) or isinstance(obj, BaseValueType):
return str(obj)
if is_obj_type(obj, bool):
raise DataValidationError(
"Expected str, found bool. Be careful, that YAML parsers consider even"
' "no" and "yes" as a bool. Search for the Norway Problem for more'
" details. And please use quotes explicitly.",
object_path,
)
raise DataValidationError(
f"expected str (or number that would be cast to string), but found type {type(obj)}", object_path
)
def _create_int(self, obj: Any, object_path: str) -> int:
# we don't want to make an int out of anything else than other int
# except for BaseValueType class instances
if is_obj_type(obj, int) or isinstance(obj, BaseValueType):
return int(obj)
raise DataValidationError(f"expected int, found {type(obj)}", object_path)
def _create_union(self, tp: Type[T], obj: Any, object_path: str) -> T:
variants = get_generic_type_arguments(tp)
errs: List[DataValidationError] = []
for v in variants:
try:
return self.map_object(v, obj, object_path=object_path)
except DataValidationError as e:
errs.append(e)
raise DataValidationError("could not parse any of the possible variants", object_path, child_exceptions=errs)
def _create_optional(self, tp: Type[Optional[T]], obj: Any, object_path: str) -> Optional[T]:
inner: Type[Any] = get_optional_inner_type(tp)
if obj is None:
return None
return self.map_object(inner, obj, object_path=object_path)
def _create_bool(self, obj: Any, object_path: str) -> bool:
if is_obj_type(obj, bool):
return obj
raise DataValidationError(f"expected bool, found {type(obj)}", object_path)
def _create_literal(self, tp: Type[Any], obj: Any, object_path: str) -> Any:
expected = get_generic_type_arguments(tp)
if obj in expected:
return obj
raise DataValidationError(f"'{obj}' does not match any of the expected values {expected}", object_path)
def _create_base_schema_object(self, tp: Type[Any], obj: Any, object_path: str) -> "BaseSchema":
if isinstance(obj, (dict, BaseSchema)):
return tp(obj, object_path=object_path)
raise DataValidationError(f"expected 'dict' or 'NoRenameBaseSchema' object, found '{type(obj)}'", object_path)
def create_value_type_object(self, tp: Type[Any], obj: Any, object_path: str) -> "BaseValueType":
if isinstance(obj, tp):
# if we already have a custom value type, just pass it through
return obj
# no validation performed, the implementation does it in the constuctor
try:
return tp(obj, object_path=object_path)
except ValueError as e:
if len(e.args) > 0 and isinstance(e.args[0], str):
msg = e.args[0]
else:
msg = f"Failed to validate value against {tp} type"
raise DataValidationError(msg, object_path) from e
def _create_default(self, obj: Any) -> Any:
if isinstance(obj, _LazyDefault):
return obj.instantiate() # type: ignore
return obj
def map_object( # noqa: PLR0911, PLR0912
self,
tp: Type[Any],
obj: Any,
default: Any = ...,
use_default: bool = False,
object_path: str = "/",
) -> Any:
"""
Given an expected type `cls` and a value object `obj`, return a new object of the given type and map fields of `obj` into it. During the mapping procedure,
runtime type checking is performed.
"""
# Disabling these checks, because I think it's much more readable as a single function
# and it's not that large at this point. If it got larger, then we should definitely split it
# pylint: disable=too-many-branches,too-many-locals,too-many-statements
# default values
if obj is None and use_default:
return self._create_default(default)
# NoneType
if is_none_type(tp):
if obj is None:
return None
raise DataValidationError(f"expected None, found '{obj}'.", object_path)
# Optional[T] (could be technically handled by Union[*variants], but this way we have better error reporting)
if is_optional(tp):
return self._create_optional(tp, obj, object_path)
# Union[*variants]
if is_union(tp):
return self._create_union(tp, obj, object_path)
# after this, there is no place for a None object
if obj is None:
raise DataValidationError(f"unexpected value 'None' for type {tp}", object_path)
# int
if tp is int:
return self._create_int(obj, object_path)
# str
if tp is str:
return self._create_str(obj, object_path)
# bool
if tp is bool:
return self._create_bool(obj, object_path)
# float
if tp is float:
raise NotImplementedError(
"Floating point values are not supported in the object mapper."
" Please implement them and be careful with type coercions"
)
# Literal[T]
if is_literal(tp):
return self._create_literal(tp, obj, object_path)
# Dict[K,V]
if is_dict(tp):
return self._create_dict(tp, obj, object_path)
# any Enums (probably used only internally in DataValidator)
if is_enum(tp):
if isinstance(obj, tp):
return obj
raise DataValidationError(f"unexpected value '{obj}' for enum '{tp}'", object_path)
# List[T]
if is_list(tp):
return self._create_list(tp, obj, object_path)
# Tuple[A,B,C,D,...]
if is_tuple(tp):
return self._create_tuple(tp, obj, object_path)
# type of obj and cls type match
if is_obj_type(obj, tp):
return obj
# when the specified type is Any, just return the given value
# on mypy version 1.11.0 comparison-overlap error started popping up
# https://github.com/python/mypy/issues/17665
if tp == Any: # type: ignore[comparison-overlap]
return obj
# BaseValueType subclasses
if inspect.isclass(tp) and issubclass(tp, BaseValueType):
return self.create_value_type_object(tp, obj, object_path)
# BaseGenericTypeWrapper subclasses
if 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
if inspect.isclass(tp) and issubclass(tp, BaseSchema):
return self._create_base_schema_object(tp, obj, object_path)
# if the object matches, just pass it through
if inspect.isclass(tp) and isinstance(obj, tp):
return obj
# default error handler
raise DataValidationError(
f"Type {tp} cannot be parsed. This is a implementation error. "
"Please fix your types in the class or improve the parser/validator.",
object_path,
)
def is_obj_type_valid(self, obj: Any, tp: Type[Any]) -> bool:
"""
Runtime type checking. Validate, that a given object is of a given type.
"""
try:
self.map_object(tp, obj)
return True
except (DataValidationError, ValueError):
return False
def _assign_default(self, obj: Any, name: str, python_type: Any, object_path: str) -> None:
cls = obj.__class__
try:
default = self._create_default(getattr(cls, name, None))
except ValueError as e:
raise DataValidationError(str(e), f"{object_path}/{name}") from e
value = self.map_object(python_type, default, object_path=f"{object_path}/{name}")
setattr(obj, name, value)
def _assign_field(self, obj: Any, name: str, python_type: Any, value: Any, object_path: str) -> None:
value = self.map_object(python_type, value, object_path=f"{object_path}/{name}")
setattr(obj, name, value)
def _assign_fields(self, obj: Any, source: Union[Dict[str, Any], "BaseSchema", None], object_path: str) -> Set[str]:
"""
Order of assignment:
1. all direct assignments
2. assignments with conversion method
"""
cls = obj.__class__
annot = cls.__dict__.get("__annotations__", {})
errs: List[DataValidationError] = []
used_keys: Set[str] = set()
for name, python_type in annot.items():
try:
if is_internal_field_name(name):
continue
# populate field
if source is None:
self._assign_default(obj, name, python_type, object_path)
# check for invalid configuration with both transformation function and default value
elif hasattr(obj, f"_{name}") and hasattr(obj, name):
raise RuntimeError(
f"Field '{obj.__class__.__name__}.{name}' has default value and transformation function at"
" the same time. That is now allowed. Store the default in the transformation function."
)
# there is a transformation function to create the value
elif hasattr(obj, f"_{name}") and callable(getattr(obj, f"_{name}")):
val = self._get_converted_value(obj, name, source, object_path)
self._assign_field(obj, name, python_type, val, object_path)
used_keys.add(name)
# source just contains the value
elif name in source:
val = source[name]
self._assign_field(obj, name, python_type, val, object_path)
used_keys.add(name)
# there is a default value, or the type is optional => store the default or null
elif hasattr(obj, name) or is_optional(python_type):
self._assign_default(obj, name, python_type, object_path)
# we expected a value but it was not there
else:
errs.append(DataValidationError(f"missing attribute '{name}'.", object_path))
except DataValidationError as e:
errs.append(e)
if len(errs) == 1:
raise errs[0]
if len(errs) > 1:
raise AggregateDataValidationError(object_path, errs)
return used_keys
def _get_converted_value(self, obj: Any, key: str, source: TSource, object_path: str) -> Any:
"""
Get a value of a field by invoking appropriate transformation function.
"""
try:
func = getattr(obj.__class__, f"_{key}")
argc = len(inspect.signature(func).parameters)
if argc == 1:
# it is a static method
return func(source)
if argc == 2:
# it is a instance method
return func(_create_untouchable("obj"), source)
raise RuntimeError("Transformation function has wrong number of arguments")
except ValueError as e:
if len(e.args) > 0 and isinstance(e.args[0], str):
msg = e.args[0]
else:
msg = "Failed to validate value type"
raise DataValidationError(msg, object_path) from e
def object_constructor(self, obj: Any, source: Union["BaseSchema", Dict[Any, Any]], object_path: str) -> None:
"""
Delegated constructor for the NoRenameBaseSchema class.
The reason this method is delegated to the mapper is due to renaming. Like this, we don't have to
worry about a different BaseSchema class, when we want to have dynamically renamed fields.
"""
# As this is a delegated constructor, we must ignore protected access warnings
# sanity check
if not isinstance(source, (BaseSchema, dict)): # type: ignore
raise DataValidationError(f"expected dict-like object, found '{type(source)}'", object_path)
# construct lower level schema first if configured to do so
if obj._LAYER is not None: # noqa: SLF001
source = obj._LAYER(source, object_path=object_path) # pylint: disable=not-callable # noqa: SLF001
# assign fields
used_keys = self._assign_fields(obj, source, object_path)
# check for unused keys in the source object
if source and not isinstance(source, BaseSchema):
unused = source.keys() - used_keys
if len(unused) > 0:
keys = ", ".join((f"'{u}'" for u in unused))
raise DataValidationError(
f"unexpected extra key(s) {keys}",
object_path,
)
# validate the constructed value
try:
obj._validate() # noqa: SLF001
except ValueError as e:
raise DataValidationError(e.args[0] if len(e.args) > 0 else "Validation error", object_path or "/") from e
class BaseSchema(Serializable):
"""
Base class for modeling configuration schema. It somewhat resembles standard dataclasses with additional
functionality:
* type validation
* data conversion
To create an instance of this class, you have to provide source data in the form of dict-like object.
Generally, raw dict or another `BaseSchema` instance. The provided data object is traversed, transformed
and validated before assigned to the appropriate fields (attributes).
Fields (attributes)
===================
The fields (or attributes) of the class are defined the same way as in a dataclass by creating a class-level
type-annotated fields. An example of that is:
class A(BaseSchema):
awesome_number: int
If your `BaseSchema` instance has a field with type of a BaseSchema, its value is recursively created
from the nested input data. This way, you can specify a complex tree of BaseSchema's and use the root
BaseSchema to create instance of everything.
Transformation
==============
You can provide the BaseSchema class with a field and a function with the same name, but starting with
underscore ('_'). For example, you could have field called `awesome_number` and function called
`_awesome_number(self, source)`. The function takes one argument - the source data (optionally with self,
but you are not supposed to touch that). It can read any data from the source object and return a value of
an appropriate type, which will be assigned to the field `awesome_number`. If you want to report an error
during validation, raise a `ValueError` exception.
Using this, you can convert any input values into any type and field you want. To make the conversion easier
to write, you could also specify a special class variable called `_LAYER` pointing to another
BaseSchema class. This causes the source object to be first parsed as the specified additional layer of BaseSchema and after that
used a source for this class. This therefore allows nesting of transformation functions.
Validation
==========
All assignments to fields during object construction are checked at runtime for proper types. This means,
you are free to use an untrusted source object and turn it into a data structure, where you are sure what
is what.
You can also define a `_validate` method, which will be called once the whole data structure is built. You
can validate the data in there and raise a `ValueError`, if they are invalid.
Default values
==============
If you create a field with a value, it will be used as a default value whenever the data in source object
are not present. As a special case, default value for Optional type is None if not specified otherwise. You
are not allowed to have a field with a default value and a transformation function at once.
Example
=======
See tests/utils/test_modelling.py for example usage.
"""
_LAYER: Optional[Type["BaseSchema"]] = None
_MAPPER: ObjectMapper = ObjectMapper()
def __init__(self, source: TSource = None, object_path: str = ""): # pylint: disable=[super-init-not-called]
# save source data (and drop information about nullness)
source = source or {}
self.__source: Union[Dict[str, Any], BaseSchema] = source
# delegate the rest of the constructor
self._MAPPER.object_constructor(self, source, object_path)
def get_unparsed_data(self) -> Dict[str, Any]:
if isinstance(self.__source, BaseSchema):
return self.__source.get_unparsed_data()
if isinstance(self.__source, Renamed):
return self.__source.original()
return self.__source
def __getitem__(self, key: str) -> Any:
if not hasattr(self, key):
raise RuntimeError(f"Object '{self}' of type '{type(self)}' does not have field named '{key}'")
return getattr(self, key)
def __contains__(self, item: Any) -> bool:
return hasattr(self, item)
def _validate(self) -> None:
"""
Validation procedure called after all field are assigned. Should throw a ValueError in case of failure.
"""
def __eq__(self, o: object) -> bool:
cls = self.__class__
if not isinstance(o, cls):
return False
annot = cls.__dict__.get("__annotations__", {})
for name in annot.keys():
if getattr(self, name) != getattr(o, name):
return False
return True
@classmethod
def json_schema(
cls: Type["BaseSchema"],
schema_id: Optional[str] = None,
title: Optional[str] = None,
description: Optional[str] = None,
include_schema_definition: bool = True,
) -> Dict[Any, Any]:
if cls._LAYER is not None:
return cls._LAYER.json_schema(
schema_id=schema_id,
title=title,
description=description,
include_schema_definition=include_schema_definition,
)
schema: Dict[Any, Any] = {}
if include_schema_definition:
schema["$schema"] = "https://json-schema.org/draft/2020-12/schema"
if schema_id is not None:
schema["$id"] = schema_id
if title is not None:
schema["title"] = title
if description is not None:
schema["description"] = description
elif cls.__doc__ is not None:
schema["description"] = _split_docstring(cls.__doc__)[0]
schema["type"] = "object"
schema["properties"] = _get_properties_schema(cls)
return schema
def to_dict(self) -> Dict[Any, Any]:
res: Dict[Any, Any] = {}
cls = self.__class__
annot = cls.__dict__.get("__annotations__", {})
for name in annot:
res[name] = Serializable.serialize(getattr(self, name))
return res
class RenamingObjectMapper(ObjectMapper):
"""
Same as object mapper, but it uses collection wrappers from the module `renamed` to perform dynamic field renaming.
More specifically:
- it renames all properties in (nested) objects
- it does not rename keys in dictionaries
"""
def _create_dict(self, tp: Type[Any], obj: Dict[Any, Any], object_path: str) -> Dict[Any, Any]:
if isinstance(obj, Renamed):
obj = obj.original()
return super()._create_dict(tp, obj, object_path)
def _create_base_schema_object(self, tp: Type[Any], obj: Any, object_path: str) -> "BaseSchema":
if isinstance(obj, dict):
obj = renamed(obj)
return super()._create_base_schema_object(tp, obj, object_path)
def object_constructor(self, obj: Any, source: Union["BaseSchema", Dict[Any, Any]], object_path: str) -> None:
if isinstance(source, dict):
source = renamed(source)
return super().object_constructor(obj, source, object_path)
# export as a standalone functions for simplicity compatibility
is_obj_type_valid = ObjectMapper().is_obj_type_valid
map_object = ObjectMapper().map_object
class ConfigSchema(BaseSchema):
"""
Same as BaseSchema, but maps with RenamingObjectMapper
"""
_MAPPER: ObjectMapper = RenamingObjectMapper()
|