Coverage for pydantic/_internal/_schema_generation_shared.py: 90.91%

1"""Types and utility functions used by various other internal tools.""" 

2 

3from __future__ import annotations 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

4 

5from typing import TYPE_CHECKING, Any, Callable 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

6 

7from pydantic_core import core_schema 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

8from typing_extensions import Literal 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

9 

10from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

11 

12if TYPE_CHECKING: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

13 from ..json_schema import GenerateJsonSchema, JsonSchemaValue 

14 from ._core_utils import CoreSchemaOrField 

15 from ._generate_schema import GenerateSchema 

16 

17 GetJsonSchemaFunction = Callable[[CoreSchemaOrField, GetJsonSchemaHandler], JsonSchemaValue] 

18 HandlerOverride = Callable[[CoreSchemaOrField], JsonSchemaValue] 

19 

20 

21class GenerateJsonSchemaHandler(GetJsonSchemaHandler): 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

22 """JsonSchemaHandler implementation that doesn't do ref unwrapping by default. 

23 

24 This is used for any Annotated metadata so that we don't end up with conflicting 

25 modifications to the definition schema. 

26 

27 Used internally by Pydantic, please do not rely on this implementation. 

28 See `GetJsonSchemaHandler` for the handler API. 

29 """ 

30 

31 def __init__(self, generate_json_schema: GenerateJsonSchema, handler_override: HandlerOverride | None) -> None: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

32 self.generate_json_schema = generate_json_schema 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

33 self.handler = handler_override or generate_json_schema.generate_inner 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

34 self.mode = generate_json_schema.mode 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

35 

36 def __call__(self, core_schema: CoreSchemaOrField, /) -> JsonSchemaValue: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

37 return self.handler(core_schema) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

38 

39 def resolve_ref_schema(self, maybe_ref_json_schema: JsonSchemaValue) -> JsonSchemaValue: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

40 """Resolves `$ref` in the json schema. 

41 

42 This returns the input json schema if there is no `$ref` in json schema. 

43 

44 Args: 

45 maybe_ref_json_schema: The input json schema that may contains `$ref`. 

46 

47 Returns: 

48 Resolved json schema. 

49 

50 Raises: 

51 LookupError: If it can't find the definition for `$ref`. 

52 """ 

53 if '$ref' not in maybe_ref_json_schema: 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

54 return maybe_ref_json_schema 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

55 ref = maybe_ref_json_schema['$ref'] 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

56 json_schema = self.generate_json_schema.get_schema_from_definitions(ref) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

57 if json_schema is None: 57 ↛ 58line 57 didn't jump to line 58, because the condition on line 57 was never true32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

58 raise LookupError( 

59 f'Could not find a ref for {ref}.' 

60 ' Maybe you tried to call resolve_ref_schema from within a recursive model?' 

61 ) 

62 return json_schema 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

63 

64 

65class CallbackGetCoreSchemaHandler(GetCoreSchemaHandler): 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

66 """Wrapper to use an arbitrary function as a `GetCoreSchemaHandler`. 

67 

68 Used internally by Pydantic, please do not rely on this implementation. 

69 See `GetCoreSchemaHandler` for the handler API. 

70 """ 

71 

72 def __init__( 39 ctx1abcdefghijklmnopqrstGHIJKLMNOuvwxyzABCD

73 self, 

74 handler: Callable[[Any], core_schema.CoreSchema], 

75 generate_schema: GenerateSchema, 

76 ref_mode: Literal['to-def', 'unpack'] = 'to-def', 

77 ) -> None: 

78 self._handler = handler 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

79 self._generate_schema = generate_schema 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

80 self._ref_mode = ref_mode 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

81 

82 def __call__(self, source_type: Any, /) -> core_schema.CoreSchema: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

83 schema = self._handler(source_type) 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

84 ref = schema.get('ref') 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

85 if self._ref_mode == 'to-def': 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

86 if ref is not None: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

87 self._generate_schema.defs.definitions[ref] = schema 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

88 return core_schema.definition_reference_schema(ref) 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

89 return schema 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

90 else: # ref_mode = 'unpack 

91 return self.resolve_ref_schema(schema) 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

92 

93 def _get_types_namespace(self) -> dict[str, Any] | None: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

94 return self._generate_schema._types_namespace 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

95 

96 def generate_schema(self, source_type: Any, /) -> core_schema.CoreSchema: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

97 return self._generate_schema.generate_schema(source_type) 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

98 

99 @property 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

100 def field_name(self) -> str | None: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

101 return self._generate_schema.field_name_stack.get() 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

102 

103 def resolve_ref_schema(self, maybe_ref_schema: core_schema.CoreSchema) -> core_schema.CoreSchema: 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

104 """Resolves reference in the core schema. 

105 

106 Args: 

107 maybe_ref_schema: The input core schema that may contains reference. 

108 

109 Returns: 

110 Resolved core schema. 

111 

112 Raises: 

113 LookupError: If it can't find the definition for reference. 

114 """ 

115 if maybe_ref_schema['type'] == 'definition-ref': 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

116 ref = maybe_ref_schema['schema_ref'] 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

117 if ref not in self._generate_schema.defs.definitions: 117 ↛ 118line 117 didn't jump to line 118, because the condition on line 117 was never true41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

118 raise LookupError( 

119 f'Could not find a ref for {ref}.' 

120 ' Maybe you tried to call resolve_ref_schema from within a recursive model?' 

121 ) 

122 return self._generate_schema.defs.definitions[ref] 41 ctx1abcdefghijEFklmnopqrstGHIJKLMNOuvwxyzABCD

123 elif maybe_ref_schema['type'] == 'definitions': 123 ↛ 124line 123 didn't jump to line 124, because the condition on line 123 was never true32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD

124 return self.resolve_ref_schema(maybe_ref_schema['schema']) 

125 return maybe_ref_schema 32 ctx1abcdefghijEFklmnopqrstuvwxyzABCD