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 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

4 

5from typing import TYPE_CHECKING, Any, Callable 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

6 

7from pydantic_core import core_schema 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

8from typing_extensions import Literal 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

9 

10from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

11 

12if TYPE_CHECKING: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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): 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

32 self.generate_json_schema = generate_json_schema 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

33 self.handler = handler_override or generate_json_schema.generate_inner 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

34 self.mode = generate_json_schema.mode 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

35 

36 def __call__(self, core_schema: CoreSchemaOrField, /) -> JsonSchemaValue: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

37 return self.handler(core_schema) 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

38 

39 def resolve_ref_schema(self, maybe_ref_json_schema: JsonSchemaValue) -> JsonSchemaValue: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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: 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

54 return maybe_ref_json_schema 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

55 ref = maybe_ref_json_schema['$ref'] 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

56 json_schema = self.generate_json_schema.get_schema_from_definitions(ref) 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

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

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 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

63 

64 

65class CallbackGetCoreSchemaHandler(GetCoreSchemaHandler): 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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__( 46 ctx1abcdefghijklmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

79 self._generate_schema = generate_schema 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

80 self._ref_mode = ref_mode 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

81 

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

83 schema = self._handler(source_type) 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

84 ref = schema.get('ref') 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

85 if self._ref_mode == 'to-def': 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

86 if ref is not None: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

87 self._generate_schema.defs.definitions[ref] = schema 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

88 return core_schema.definition_reference_schema(ref) 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

89 return schema 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

90 else: # ref_mode = 'unpack 

91 return self.resolve_ref_schema(schema) 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

92 

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

94 return self._generate_schema._types_namespace 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

95 

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

97 return self._generate_schema.generate_schema(source_type) 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

98 

99 @property 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

100 def field_name(self) -> str | None: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

101 return self._generate_schema.field_name_stack.get() 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ

102 

103 def resolve_ref_schema(self, maybe_ref_schema: core_schema.CoreSchema) -> core_schema.CoreSchema: 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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': 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

116 ref = maybe_ref_schema['schema_ref'] 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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 true48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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] 48 ctx1abcdefghijklKLmnopqrstuvwxMNOPQRSTUVyzABCDEFGHIJ

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

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

125 return maybe_ref_schema 38 ctx1abcdefghijklKLmnopqrstuvwxyzABCDEFGHIJ