Coverage for pydantic/plugin/_schema_validator.py: 100.00%

1"""Pluggable schema validator for pydantic.""" 

2 

3from __future__ import annotations 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

4 

5import functools 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

6from typing import TYPE_CHECKING, Any, Callable, Iterable, TypeVar 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

7 

8from pydantic_core import CoreConfig, CoreSchema, SchemaValidator, ValidationError 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

9from typing_extensions import Literal, ParamSpec 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

10 

11if TYPE_CHECKING: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

12 from . import BaseValidateHandlerProtocol, PydanticPluginProtocol, SchemaKind, SchemaTypePath 

13 

14 

15P = ParamSpec('P') 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

16R = TypeVar('R') 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

17Event = Literal['on_validate_python', 'on_validate_json', 'on_validate_strings'] 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

18events: list[Event] = list(Event.__args__) # type: ignore 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

19 

20 

21def create_schema_validator( 39 ctx1abcdefghijklmnopqrstGHIJKLMNOuvwxyzABCD

22 schema: CoreSchema, 

23 schema_type: Any, 

24 schema_type_module: str, 

25 schema_type_name: str, 

26 schema_kind: SchemaKind, 

27 config: CoreConfig | None = None, 

28 plugin_settings: dict[str, Any] | None = None, 

29) -> SchemaValidator: 

30 """Create a `SchemaValidator` or `PluggableSchemaValidator` if plugins are installed. 

31 

32 Returns: 

33 If plugins are installed then return `PluggableSchemaValidator`, otherwise return `SchemaValidator`. 

34 """ 

35 from . import SchemaTypePath 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

36 from ._loader import get_plugins 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

37 

38 plugins = get_plugins() 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

39 if plugins: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

40 return PluggableSchemaValidator( 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

41 schema, 

42 schema_type, 

43 SchemaTypePath(schema_type_module, schema_type_name), 

44 schema_kind, 

45 config, 

46 plugins, 

47 plugin_settings or {}, 

48 ) # type: ignore 

49 else: 

50 return SchemaValidator(schema, config) 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

51 

52 

53class PluggableSchemaValidator: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

54 """Pluggable schema validator.""" 

55 

56 __slots__ = '_schema_validator', 'validate_json', 'validate_python', 'validate_strings' 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

57 

58 def __init__( 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

59 self, 

60 schema: CoreSchema, 

61 schema_type: Any, 

62 schema_type_path: SchemaTypePath, 

63 schema_kind: SchemaKind, 

64 config: CoreConfig | None, 

65 plugins: Iterable[PydanticPluginProtocol], 

66 plugin_settings: dict[str, Any], 

67 ) -> None: 

68 self._schema_validator = SchemaValidator(schema, config) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

69 

70 python_event_handlers: list[BaseValidateHandlerProtocol] = [] 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

71 json_event_handlers: list[BaseValidateHandlerProtocol] = [] 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

72 strings_event_handlers: list[BaseValidateHandlerProtocol] = [] 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

73 for plugin in plugins: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

74 try: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

75 p, j, s = plugin.new_schema_validator( 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

76 schema, schema_type, schema_type_path, schema_kind, config, plugin_settings 

77 ) 

78 except TypeError as e: # pragma: no cover 

79 raise TypeError(f'Error using plugin `{plugin.__module__}:{plugin.__class__.__name__}`: {e}') from e 

80 if p is not None: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

81 python_event_handlers.append(p) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

82 if j is not None: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

83 json_event_handlers.append(j) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

84 if s is not None: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

85 strings_event_handlers.append(s) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

86 

87 self.validate_python = build_wrapper(self._schema_validator.validate_python, python_event_handlers) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

88 self.validate_json = build_wrapper(self._schema_validator.validate_json, json_event_handlers) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

89 self.validate_strings = build_wrapper(self._schema_validator.validate_strings, strings_event_handlers) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

90 

91 def __getattr__(self, name: str) -> Any: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

92 return getattr(self._schema_validator, name) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

93 

94 

95def build_wrapper(func: Callable[P, R], event_handlers: list[BaseValidateHandlerProtocol]) -> Callable[P, R]: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

96 if not event_handlers: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

97 return func 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

98 else: 

99 on_enters = tuple(h.on_enter for h in event_handlers if filter_handlers(h, 'on_enter')) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

100 on_successes = tuple(h.on_success for h in event_handlers if filter_handlers(h, 'on_success')) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

101 on_errors = tuple(h.on_error for h in event_handlers if filter_handlers(h, 'on_error')) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

102 on_exceptions = tuple(h.on_exception for h in event_handlers if filter_handlers(h, 'on_exception')) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

103 

104 @functools.wraps(func) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

105 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

106 for on_enter_handler in on_enters: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

107 on_enter_handler(*args, **kwargs) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

108 

109 try: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

110 result = func(*args, **kwargs) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

111 except ValidationError as error: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

112 for on_error_handler in on_errors: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

113 on_error_handler(error) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

114 raise 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

115 except Exception as exception: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

116 for on_exception_handler in on_exceptions: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

117 on_exception_handler(exception) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

118 raise 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

119 else: 

120 for on_success_handler in on_successes: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

121 on_success_handler(result) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

122 return result 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

123 

124 return wrapper 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

125 

126 

127def filter_handlers(handler_cls: BaseValidateHandlerProtocol, method_name: str) -> bool: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

128 """Filter out handler methods which are not implemented by the plugin directly - e.g. are missing 

129 or are inherited from the protocol. 

130 """ 

131 handler = getattr(handler_cls, method_name, None) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

132 if handler is None: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

133 return False 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

134 elif handler.__module__ == 'pydantic.plugin': 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

135 # this is the original handler, from the protocol due to runtime inheritance 

136 # we don't want to call it 

137 return False 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

138 else: 

139 return True 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD