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

69 statements  

« prev     ^ index     » next       coverage.py v7.5.4, created at 2024-07-03 19:29 +0000

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

2 

3from __future__ import annotations 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

4 

5import functools 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

6from typing import TYPE_CHECKING, Any, Callable, Iterable, TypeVar 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

7 

8from pydantic_core import CoreConfig, CoreSchema, SchemaValidator, ValidationError 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

9from typing_extensions import Literal, ParamSpec 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

10 

11if TYPE_CHECKING: 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

12 from . import BaseValidateHandlerProtocol, PydanticPluginProtocol, SchemaKind, SchemaTypePath 

13 

14 

15P = ParamSpec('P') 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

16R = TypeVar('R') 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

17Event = Literal['on_validate_python', 'on_validate_json', 'on_validate_strings'] 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

18events: list[Event] = list(Event.__args__) # type: ignore 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

19 

20 

21def create_schema_validator( 1abcdefghijklmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

36 from ._loader import get_plugins 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

37 

38 plugins = get_plugins() 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

39 if plugins: 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

40 return PluggableSchemaValidator( 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

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) 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

51 

52 

53class PluggableSchemaValidator: 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

54 """Pluggable schema validator.""" 

55 

56 __slots__ = '_schema_validator', 'validate_json', 'validate_python', 'validate_strings' 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

57 

58 def __init__( 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

69 

70 python_event_handlers: list[BaseValidateHandlerProtocol] = [] 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

71 json_event_handlers: list[BaseValidateHandlerProtocol] = [] 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

72 strings_event_handlers: list[BaseValidateHandlerProtocol] = [] 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

73 for plugin in plugins: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

74 try: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

75 p, j, s = plugin.new_schema_validator( 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

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: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

81 python_event_handlers.append(p) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

82 if j is not None: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

83 json_event_handlers.append(j) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

84 if s is not None: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

85 strings_event_handlers.append(s) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

86 

87 self.validate_python = build_wrapper(self._schema_validator.validate_python, python_event_handlers) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

88 self.validate_json = build_wrapper(self._schema_validator.validate_json, json_event_handlers) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

89 self.validate_strings = build_wrapper(self._schema_validator.validate_strings, strings_event_handlers) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

90 

91 def __getattr__(self, name: str) -> Any: 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

92 return getattr(self._schema_validator, name) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

93 

94 

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

96 if not event_handlers: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

97 return func 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

98 else: 

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

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

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

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

103 

104 @functools.wraps(func) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

105 def wrapper(*args: P.args, **kwargs: P.kwargs) -> R: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

106 for on_enter_handler in on_enters: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

107 on_enter_handler(*args, **kwargs) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

108 

109 try: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

110 result = func(*args, **kwargs) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

111 except ValidationError as error: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

112 for on_error_handler in on_errors: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

113 on_error_handler(error) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

114 raise 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

115 except Exception as exception: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

116 for on_exception_handler in on_exceptions: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

117 on_exception_handler(exception) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

118 raise 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

119 else: 

120 for on_success_handler in on_successes: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

121 on_success_handler(result) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

122 return result 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

123 

124 return wrapper 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

125 

126 

127def filter_handlers(handler_cls: BaseValidateHandlerProtocol, method_name: str) -> bool: 1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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) 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

132 if handler is None: 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

133 return False 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

134 elif handler.__module__ == 'pydantic.plugin': 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

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

136 # we don't want to call it 

137 return False 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

138 else: 

139 return True 1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ