Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/pydantic/_internal/_schema_generation_shared.py: 50%

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

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from typing import TYPE_CHECKING, Any, Literal 

7 

8from pydantic_core import core_schema 

9 

10from ..annotated_handlers import GetCoreSchemaHandler, GetJsonSchemaHandler 

11 

12if TYPE_CHECKING: 

13 from ..json_schema import GenerateJsonSchema, JsonSchemaValue 

14 from ._core_utils import CoreSchemaOrField 

15 from ._generate_schema import GenerateSchema 

16 from ._namespace_utils import NamespacesTuple 

17 

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

19 HandlerOverride = Callable[[CoreSchemaOrField], JsonSchemaValue] 

20 

21 

22class GenerateJsonSchemaHandler(GetJsonSchemaHandler): 

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

24 

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

26 modifications to the definition schema. 

27 

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

29 See `GetJsonSchemaHandler` for the handler API. 

30 """ 

31 

32 def __init__(self, generate_json_schema: GenerateJsonSchema, handler_override: HandlerOverride | None) -> None: 

33 self.generate_json_schema = generate_json_schema 

34 self.handler = handler_override or generate_json_schema.generate_inner 

35 self.mode = generate_json_schema.mode 

36 

37 def __call__(self, core_schema: CoreSchemaOrField, /) -> JsonSchemaValue: 

38 return self.handler(core_schema) 

39 

40 def resolve_ref_schema(self, maybe_ref_json_schema: JsonSchemaValue) -> JsonSchemaValue: 

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

42 

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

44 

45 Args: 

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

47 

48 Returns: 

49 Resolved json schema. 

50 

51 Raises: 

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

53 """ 

54 if '$ref' not in maybe_ref_json_schema: 

55 return maybe_ref_json_schema 

56 ref = maybe_ref_json_schema['$ref'] 

57 json_schema = self.generate_json_schema.get_schema_from_definitions(ref) 

58 if json_schema is None: 

59 raise LookupError( 

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

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

62 ) 

63 return json_schema 

64 

65 

66class CallbackGetCoreSchemaHandler(GetCoreSchemaHandler): 

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

68 

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

70 See `GetCoreSchemaHandler` for the handler API. 

71 """ 

72 

73 def __init__( 

74 self, 

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

76 generate_schema: GenerateSchema, 

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

78 ) -> None: 

79 self._handler = handler 

80 self._generate_schema = generate_schema 

81 self._ref_mode = ref_mode 

82 

83 def __call__(self, source_type: Any, /) -> core_schema.CoreSchema: 

84 schema = self._handler(source_type) 

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

86 ref = schema.get('ref') 

87 if ref is not None: 

88 return self._generate_schema.defs.create_definition_reference_schema(schema) 

89 return schema 

90 else: # ref_mode = 'unpack' 

91 return self.resolve_ref_schema(schema) 

92 

93 def _get_types_namespace(self) -> NamespacesTuple: 

94 return self._generate_schema._types_namespace 

95 

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

97 return self._generate_schema.generate_schema(source_type) 

98 

99 @property 

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

101 return self._generate_schema.field_name_stack.get() 

102 

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

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

116 ref = maybe_ref_schema['schema_ref'] 

117 definition = self._generate_schema.defs.get_schema_from_ref(ref) 

118 if definition is None: 

119 raise LookupError( 

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

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

122 ) 

123 return definition 

124 elif maybe_ref_schema['type'] == 'definitions': 

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

126 return maybe_ref_schema