Coverage for src/metador_core/schema/parser.py: 100%
48 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-02 09:33 +0000
« prev ^ index » next coverage.py v7.3.2, created at 2023-11-02 09:33 +0000
1"""Simplify creation of custom parsers for pydantic models."""
2from typing import Any, ClassVar, Dict, Type, TypeVar
4from pydantic import BaseModel
6T = TypeVar("T")
9class BaseParser:
10 """Parsers that work with the ParserMixin must inherit from this class."""
12 schema_info: Dict[str, Any] = {}
13 strict: bool = True
15 @classmethod
16 def parse(cls, target: Type[T], v: Any) -> T:
17 """Override and implement this method for custom parsing.
19 The default implementation will simply pass through
20 any instances of `tcls` unchanged and fail on anything else.
22 Make sure that the parser can also handle any object that itself
23 produces as an input.
25 By default, parsers are expected to normalize the input,
26 i.e. produce an instance of `tcls`, any other returned type
27 will lead to an exception.
29 If you know what you are doing, set `strict=False` to
30 disable this behavior.
32 Args:
33 target: class the value should be parsed into
34 v: value to be parsed
35 """
36 if target is not None and not isinstance(v, target):
37 raise TypeError(f"Expected {target.__name__}, but got {type(v).__name__}!")
38 return v
41def run_parser(cls: Type[BaseParser], target: Type[T], value: Any):
42 """Parse and validate passed value."""
43 # print("call parser", cls, "from", tcls, "on", value, ":", type(value))
44 ret = cls.parse(target, value)
45 if cls.strict and not isinstance(ret, target):
46 msg = f"Parser returned: {type(ret).__name__}, "
47 msg += f"expected: {target.__name__} (strict=True)"
48 raise RuntimeError(msg)
49 return ret
52def get_parser(cls):
53 """Return inner Parser class, or None.
55 If the inner Parser class is not a subclass of `BaseParser`,
56 will raise an exception, as this is most likely an error.
57 """
58 if parser := cls.__dict__.get("Parser"):
59 if not issubclass(parser, BaseParser):
60 msg = f"{cls}: {cls.Parser.__name__} must be a subclass of {BaseParser.__name__}!"
61 raise TypeError(msg)
62 return parser
65class NoParserDefined:
66 ...
69class ParserMixin:
70 """Mixin class to simplify creation of custom pydantic field types.
72 Can also be mixed into arbitrary classes, not just pydantic models,
73 that is why it is kept separately from the top level base model we use.
75 Also, we avoid using a custom metaclass for the mixin itself,
76 to increase compatibility with various classes.
77 """
79 Parser: ClassVar[Type[BaseParser]]
81 @classmethod
82 def __get_validators__(cls):
83 pfunc = cls.__dict__.get("__parser_func__")
84 if pfunc is None:
85 if parser := get_parser(cls):
87 def wrapper_func(cls, value, values=None, config=None, field=None):
88 return run_parser(parser, field.type_, value)
90 pfunc = wrapper_func
91 else:
92 pfunc = NoParserDefined
93 cls.__parser_func__ = pfunc # cache it
95 if pfunc is not NoParserDefined: # return cached parser function
96 yield pfunc
98 # important: if no parser is given
99 # and class is a model,
100 # also return the default validate function of the model!
101 if issubclass(cls, BaseModel):
102 yield cls.validate
104 @classmethod
105 def __modify_schema__(cls, schema):
106 if parser := get_parser(cls):
107 if schema_info := parser.schema_info:
108 schema.update(**schema_info)
111__all__ = ["BaseParser", "ParserMixin"]